feat: alterando api de pluviometro

pipelines/rj_cor/meteorologia/precipitacao_alertario/flows.py

@@ -1,4 +1,5 @@
 # -*- coding: utf-8 -*-
+# pylint: disable=C0103
 """
 Flows for precipitacao_alertario
 """
@@ -12,7 +13,6 @@
 from pipelines.constants import constants
 from pipelines.utils.constants import constants as utils_constants
 from pipelines.rj_cor.meteorologia.precipitacao_alertario.tasks import (
-    download,
     tratar_dados,
     salvar_dados,
 )
@@ -34,7 +34,7 @@
     ],
 ) as cor_meteorologia_precipitacao_alertario:
 
-    DATASET_ID = "meio_ambiente_clima"
+    DATASET_ID = "clima_pluviometro"
     TABLE_ID = "taxa_precipitacao_alertario"
     DUMP_MODE = "append"
 
@@ -56,9 +56,8 @@
         default=dump_to_gcs_constants.MAX_BYTES_PROCESSED_PER_TABLE.value,
     )
 
-    filename, current_time = download()
-    dados, empty_data = tratar_dados(
-        filename=filename, dataset_id=DATASET_ID, table_id=TABLE_ID
+    dados, empty_data, current_time = tratar_dados(
+        dataset_id=DATASET_ID, table_id=TABLE_ID
     )
 
     # If dataframe is empty stop flow

pipelines/rj_cor/meteorologia/precipitacao_alertario/schedules.py

@@ -1,4 +1,5 @@
 # -*- coding: utf-8 -*-
+# pylint: disable=C0103
 """
 Schedules for precipitacao_alertario
 Rodar a cada 5 minutos

pipelines/rj_cor/meteorologia/precipitacao_alertario/tasks.py

@@ -1,135 +1,69 @@
 # -*- coding: utf-8 -*-
+# pylint: disable=C0103
 """
 Tasks for precipitacao_alertario
 """
 from datetime import timedelta
-import ftplib
 import os
 from pathlib import Path
-import socket
 from typing import Union, Tuple
 
 import numpy as np
 import pandas as pd
 import pendulum
 from prefect import task
+import pandas_read_xml as pdx
+
+# from prefect import context
 
 from pipelines.constants import constants
 from pipelines.rj_cor.meteorologia.utils import save_updated_rows_on_redis
-from pipelines.utils.utils import get_vault_secret, log
+from pipelines.utils.utils import log
 
 
 @task(
-    nout=2,
+    nout=3,
     max_retries=constants.TASK_MAX_RETRIES.value,
     retry_delay=timedelta(seconds=constants.TASK_RETRY_DELAY.value),
 )
-def download() -> Tuple[pd.DataFrame, str]:
-    """
-    Faz o request e salva dados localmente
-    """
-
-    # Acessar FTP Riomidia
-    dirname = "/alertario/"
-    filename = "EstacoesMapa.txt"
-
-    # Acessar username e password
-    dicionario = get_vault_secret("riomidia")
-    host = dicionario["data"]["host"]
-    username = dicionario["data"]["username"]
-    password = dicionario["data"]["password"]
-
-    # Cria pasta para salvar arquivo de download
-    base_path = os.path.join(os.getcwd(), "data", "precipitacao_alertario", "input")
-
-    if not os.path.exists(base_path):
-        os.makedirs(base_path)
-
-    save_on = os.path.join(base_path, filename)
-
-    try:
-        ftp = ftplib.FTP(host)
-    except (socket.error, socket.gaierror):
-        log(f"ERROR: cannot reach {host}")
-        raise
-    log(f"*** Connected to host {host}")
-
-    try:
-        ftp.login(username, password)
-    except ftplib.error_perm:
-        log("ERROR: cannot login")
-        ftp.quit()
-        raise
-    log("*** Logged in successfully")
-
-    try:
-        ftp.cwd(dirname)
-    except ftplib.error_perm:
-        log(f"ERROR: cannot CD to {dirname}")
-        ftp.quit()
-        raise
-    log(f"*** Changed to folder: {dirname}")
-
-    try:
-        with open(save_on, "wb") as file:
-            log("Getting " + filename)
-            ftp.retrbinary("RETR " + filename, file.write)
-        log(f"File downloaded to {save_on}")
-    except ftplib.error_perm:
-        log(f"ERROR: cannot read file {filename}")
-        raise
-
-    ftp.quit()
-
-    # Hora atual no formato YYYYMMDDHHmm para criar partições
-    current_time = pendulum.now("America/Sao_Paulo").strftime("%Y%m%d%H%M")
-
-    return save_on, current_time
-
-
-@task(nout=2)
-def tratar_dados(
-    filename: Union[str, Path], dataset_id: str, table_id: str
-) -> Tuple[pd.DataFrame, bool]:
+def tratar_dados(dataset_id: str, table_id: str) -> Tuple[pd.DataFrame, bool]:
     """
     Renomeia colunas e filtra dados com a hora e minuto do timestamp
     de execução mais próximo à este
     """
 
-    colunas = [
-        "id_estacao",
-        "estacao",
-        "localizacao",
-        "data_medicao",
-        "latitude",
-        "longitude",
-        "acumulado_chuva_15_min",
-        "acumulado_chuva_1_h",
-        "acumulado_chuva_4_h",
-        "acumulado_chuva_24_h",
-        "acumulado_chuva_96_h",
-        "acumulado_chuva_mes",
-    ]
-
-    dados = pd.read_csv(filename, skiprows=1, sep=";", names=colunas, decimal=",")
-
-    # Adequando formato de data
-    dados["data_medicao"] = pd.to_datetime(
-        dados["data_medicao"], format="%H:%M - %d/%m/%Y"
-    )
+    # Hora atual no formato YYYYMMDDHHmm para criar partições
+    current_time = pendulum.now("America/Sao_Paulo").strftime("%Y%m%d%H%M")
 
-    # Ordenação de variáveis
-    cols_order = [
-        "data_medicao",
-        "id_estacao",
-        "acumulado_chuva_15_min",
-        "acumulado_chuva_1_h",
-        "acumulado_chuva_4_h",
-        "acumulado_chuva_24_h",
-        "acumulado_chuva_96_h",
+    url = "http://alertario.rio.rj.gov.br/upload/xml/Chuvas.xml"
+    dados = pdx.read_xml(url, ["estacoes"])
+    dados = pdx.fully_flatten(dados)
+
+    drop_cols = [
+        "@hora",
+        "estacao|@nome",
+        "estacao|@type",
+        "estacao|localizacao|@bacia",
+        "estacao|localizacao|@latitude",
+        "estacao|localizacao|@longitude",
     ]
-
-    dados = dados[cols_order]
+    rename_cols = {
+        "estacao|@id": "id_estacao",
+        "estacao|chuvas|@h01": "acumulado_chuva_1_h",
+        "estacao|chuvas|@h04": "acumulado_chuva_4_h",
+        "estacao|chuvas|@h24": "acumulado_chuva_24_h",
+        "estacao|chuvas|@h96": "acumulado_chuva_96_h",
+        "estacao|chuvas|@hora": "data_medicao_utc",
+        "estacao|chuvas|@m15": "acumulado_chuva_15_min",
+        "estacao|chuvas|@mes": "acumulado_chuva_mes",
+    }
+
+    dados = pdx.fully_flatten(dados).drop(drop_cols, axis=1).rename(rename_cols, axis=1)
+    log(f"\n[DEBUG]: df.head() {dados.head()}")
+
+    # Converte de UTC para horário São Paulo
+    dados["data_medicao_utc"] = pd.to_datetime(dados["data_medicao_utc"])
+    dados["data_medicao"] = dados["data_medicao_utc"].dt.strftime("%Y-%m-%d %H:%M:%S")
 
     # Alterando valores ND, '-' e np.nan para NULL
     dados.replace(["ND", "-", np.nan], [None, None, None], inplace=True)
@@ -144,7 +78,7 @@ def tratar_dados(
     ]
     dados[float_cols] = dados[float_cols].apply(pd.to_numeric, errors="coerce")
 
-    # Altera valores negativos para 0
+    # Altera valores negativos para None
     dados[float_cols] = np.where(dados[float_cols] < 0, None, dados[float_cols])
 
     # Elimina linhas em que o id_estacao é igual mantendo a de menor valor nas colunas float
@@ -173,8 +107,9 @@ def tratar_dados(
 
     # If df is empty stop flow
     empty_data = dados.shape[0] == 0
+    log(f"[DEBUG]: dataframe is empty: {empty_data}")
 
-    return dados, empty_data
+    return dados, empty_data, current_time
 
 
 @task