Viagens 2.0 - Validação de viagens

pipelines/flows.py

@@ -21,3 +21,4 @@
 from pipelines.migration.veiculo.flows import *  # noqa
 from pipelines.serpro.flows import *  # noqa
 from pipelines.treatment.monitoramento.flows import *  # noqa
+from pipelines.treatment.planejamento.flows import *  # noqa

pipelines/migration/br_rj_riodejaneiro_gtfs/flows.py

@@ -215,6 +215,7 @@
             + " "
             + constants.PLANEJAMENTO_MATERIALIZACAO_DATASET_ID.value,
             _vars=dbt_vars,
+            exclude="calendario",
         ).set_upstream(task=wait_captura)
 
         wait_materialize_true = update_last_captured_os(

pipelines/schedules.py

@@ -12,8 +12,11 @@
 from pipelines.constants import constants
 from pipelines.constants import constants as emd_constants
 
+cron_every_day_hour_1 = "0 1 * * *"
+cron_every_day_hour_6 = "0 6 * * *"
 cron_every_day_hour_7 = "0 7 * * *"
 cron_every_day_hour_7_minute_10 = "10 7 * * *"
+cron_every_hour_minute_6 = "6 * * * *"
 
 
 def generate_interval_schedule(

pipelines/treatment/monitoramento/CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog - monitoramento
 
+## [1.1.0] - 2024-11-08
+
+### Adicionado
+
+- Cria flow de tratamento de validação de viagens informadas (https://github.com/prefeitura-rio/pipelines_rj_smtr/pull/237)
+
 ## [1.0.0] - 2024-10-21
 
 ### Adicionado

pipelines/treatment/monitoramento/constants.py

@@ -6,7 +6,7 @@
 from datetime import datetime
 from enum import Enum
 
-from pipelines.schedules import cron_every_day_hour_7_minute_10
+from pipelines.schedules import cron_every_day_hour_6, cron_every_day_hour_7_minute_10
 from pipelines.treatment.templates.utils import DBTSelector
 
 
@@ -20,3 +20,10 @@ class constants(Enum):  # pylint: disable=c0103
         schedule_cron=cron_every_day_hour_7_minute_10,
         initial_datetime=datetime(2024, 10, 16, 0, 0, 0),
     )
+
+    VIAGEM_VALIDACAO_SELECTOR = DBTSelector(
+        name="viagem_validacao",
+        schedule_cron=cron_every_day_hour_6,
+        initial_datetime=datetime(2024, 10, 12, 0, 0, 0),
+        incremental_delay_hours=48,
+    )

pipelines/treatment/monitoramento/flows.py

@@ -9,7 +9,14 @@
     constants as rioonibus_source_constants,
 )
 from pipelines.constants import constants as smtr_constants
+from pipelines.migration.br_rj_riodejaneiro_onibus_gps_zirix.constants import (
+    constants as gps_zirix_constants,
+)
+from pipelines.schedules import cron_every_hour_minute_6
 from pipelines.treatment.monitoramento.constants import constants
+from pipelines.treatment.planejamento.constants import (
+    constants as planejamento_constants,
+)
 from pipelines.treatment.templates.flows import create_default_materialization_flow
 
 VIAGEM_INFORMADA_MATERIALIZACAO = create_default_materialization_flow(
@@ -18,3 +25,29 @@
     agent_label=smtr_constants.RJ_SMTR_AGENT_LABEL.value,
     wait=[rioonibus_source_constants.VIAGEM_INFORMADA_SOURCE.value],
 )
+
+VIAGEM_VALIDACAO_MATERIALIZACAO = create_default_materialization_flow(
+    flow_name="viagem_validacao - materializacao",
+    selector=constants.VIAGEM_VALIDACAO_SELECTOR.value,
+    agent_label=smtr_constants.RJ_SMTR_AGENT_LABEL.value,
+    wait=[
+        constants.VIAGEM_INFORMADA_SELECTOR.value,
+        planejamento_constants.PLANEJAMENTO_DIARIO_SELECTOR.value,
+        {
+            "redis_key": f"{smtr_constants.GPS_SPPO_DATASET_ID.value}\
+.{smtr_constants.GPS_SPPO_TREATED_TABLE_ID.value}",
+            "dict_key": "last_run_timestamp",
+            "datetime_format": "%Y-%m-%dT%H:%M:%S",
+            "delay_hours": smtr_constants.GPS_SPPO_MATERIALIZE_DELAY_HOURS.value,
+            "schedule_cron": cron_every_hour_minute_6,
+        },
+        {
+            "redis_key": f"{gps_zirix_constants.GPS_SPPO_ZIRIX_RAW_DATASET_ID.value}\
+.{smtr_constants.GPS_SPPO_TREATED_TABLE_ID.value}",
+            "dict_key": "last_run_timestamp",
+            "datetime_format": "%Y-%m-%dT%H:%M:%S",
+            "delay_hours": smtr_constants.GPS_SPPO_MATERIALIZE_DELAY_HOURS.value,
+            "schedule_cron": cron_every_hour_minute_6,
+        },
+    ],
+)

pipelines/treatment/planejamento/CHANGELOG.md

@@ -0,0 +1,7 @@
+# Changelog - planejamento
+
+## [1.0.0] - 2024-11-08
+
+### Adicionado
+
+- Cria flow de tratamento diário de dados de planejamento (https://github.com/prefeitura-rio/pipelines_rj_smtr/pull/237)

pipelines/treatment/planejamento/constants.py

@@ -0,0 +1,22 @@
+# -*- coding: utf-8 -*-
+"""
+Valores constantes para materialização dos dados de planejamento
+"""
+
+from datetime import datetime
+from enum import Enum
+
+from pipelines.schedules import cron_every_day_hour_1
+from pipelines.treatment.templates.utils import DBTSelector
+
+
+class constants(Enum):  # pylint: disable=c0103
+    """
+    Valores constantes para materialização dos dados de planejamento
+    """
+
+    PLANEJAMENTO_DIARIO_SELECTOR = DBTSelector(
+        name="planejamento_diario",
+        schedule_cron=cron_every_day_hour_1,
+        initial_datetime=datetime(2024, 9, 1, 0, 0, 0),
+    )

pipelines/treatment/planejamento/flows.py

@@ -0,0 +1,14 @@
+# -*- coding: utf-8 -*-
+"""
+Flows de tratamento dos dados de planejamento
+"""
+
+from pipelines.constants import constants as smtr_constants
+from pipelines.treatment.planejamento.constants import constants
+from pipelines.treatment.templates.flows import create_default_materialization_flow
+
+PLANEJAMENTO_DIARIO_MATERIALIZACAO = create_default_materialization_flow(
+    flow_name="planejamento_diario - materializacao",
+    selector=constants.PLANEJAMENTO_DIARIO_SELECTOR.value,
+    agent_label=smtr_constants.RJ_SMTR_AGENT_LABEL.value,
+)

pipelines/treatment/templates/CHANGELOG.md

@@ -1,12 +1,19 @@
 # Changelog - treatment
 
+## [1.0.2] - 2024-11-08
+
+### Alterado
+
+- Adiciona lógica para verificar fontes de dados no padrão antigo (https://github.com/prefeitura-rio/pipelines_rj_smtr/pull/237)
+
 ## [1.0.1] - 2024-10-29
 
 ### Adicionado
 
 - Adiciona as tasks `check_dbt_test_run`, `run_dbt_tests` e `dbt_data_quality_checks` (https://github.com/prefeitura-rio/pipelines_rj_smtr/pull/288)
 - Adiciona a função `parse_dbt_test_output` no `utils.py` (https://github.com/prefeitura-rio/pipelines_rj_smtr/pull/288)
 
+
 ## [1.0.0] - 2024-10-21
 
 ### Adicionado

pipelines/treatment/templates/tasks.py

@@ -8,6 +8,7 @@
 import requests
 from prefect import task
 from prefeitura_rio.pipelines_utils.logging import log
+from prefeitura_rio.pipelines_utils.redis_pal import get_redis_client
 from pytz import timezone
 
 from pipelines.constants import constants
@@ -23,7 +24,7 @@
 from pipelines.utils.gcp.bigquery import SourceTable
 from pipelines.utils.prefect import flow_is_running_local, rename_current_flow_run
 from pipelines.utils.secret import get_secret
-from pipelines.utils.utils import convert_timezone
+from pipelines.utils.utils import convert_timezone, cron_get_last_date
 
 # from pipelines.utils.utils import get_last_materialization_redis_key
 
@@ -123,7 +124,7 @@ def wait_data_sources(
     env: str,
     datetime_start: datetime,
     datetime_end: datetime,
-    data_sources: list[Union[SourceTable, DBTSelector]],
+    data_sources: list[Union[SourceTable, DBTSelector, dict]],
     skip: bool,
 ):
     """
@@ -133,7 +134,7 @@ def wait_data_sources(
         env (str): prod ou dev
         datetime_start (datetime): Datetime inicial da materialização
         datetime_end (datetime): Datetime final da materialização
-        data_sources (list[Union[SourceTable, DBTSelector]]): Fontes de dados para esperar
+        data_sources (list[Union[SourceTable, DBTSelector, dict]]): Fontes de dados para esperar
         skip (bool): se a verificação deve ser pulada ou não
     """
     if skip:
@@ -154,6 +155,22 @@ def wait_data_sources(
             elif isinstance(ds, DBTSelector):
                 name = f"{ds.name}"
                 complete = ds.is_up_to_date(env=env, timestamp=datetime_end)
+            elif isinstance(ds, dict):
+                # source dicionário utilizado para compatibilização com flows antigos
+                name = ds["redis_key"]
+                redis_client = get_redis_client()
+                last_materialization = datetime.strptime(
+                    redis_client.get(name)[ds["dict_key"]],
+                    ds["datetime_format"],
+                )
+                last_schedule = cron_get_last_date(
+                    cron_expr=ds["schedule_cron"],
+                    timestamp=datetime_end,
+                )
+                complete = convert_timezone(
+                    timestamp=last_materialization
+                ) >= last_schedule - timedelta(hours=ds.get("delay_hours", 0))
+
             else:
                 raise NotImplementedError(f"Espera por fontes do tipo {type(ds)} não implementada")
 

pipelines/treatment/templates/utils.py

@@ -72,7 +72,7 @@ def get_last_materialized_datetime(self, env: str) -> datetime:
             )
         )
 
-        return last_datetime
+        return convert_timezone(timestamp=last_datetime)
 
     def get_datetime_end(self, timestamp: datetime) -> datetime:
         """