Bilhetagem - Adiciona Transações do RioCard na passageiros_hora

pipelines/migration/br_rj_riodejaneiro_bilhetagem/CHANGELOG.md

@@ -1,5 +1,15 @@
 # Changelog - br_rj_riodejaneiro_bilhetagem
 
+## [1.3.0] - 2024-08-05
+
+### Adicionado
+
+- Cria captura da tabela `linha_consorcio` do banco de dados da Jaé (https://github.com/prefeitura-rio/pipelines_rj_smtr/pull/130)
+
+### Alterado
+
+- Renomeia flow `SMTR: Bilhetagem Transação RioCard - Materialização` para `SMTR: Bilhetagem Controle Vinculo Validador - Materialização` e adiciona modelo `transacao_riocard` no parametro `exclude`  (https://github.com/prefeitura-rio/pipelines_rj_smtr/pull/130)
+
 ## [1.2.0] - 2024-07-17
 
 ### Alterado

pipelines/migration/br_rj_riodejaneiro_bilhetagem/constants.py

@@ -432,6 +432,29 @@ class constants(Enum):  # pylint: disable=c0103
             ],  # id column to nest data on
             "interval_minutes": BILHETAGEM_TRATAMENTO_INTERVAL,
         },
+        {
+            "table_id": "linha_consorcio",
+            "partition_date_only": True,
+            "extract_params": {
+                "database": "principal_db",
+                "query": """
+                    SELECT
+                        *
+                    FROM
+                        LINHA_CONSORCIO
+                    WHERE
+                        DT_INCLUSAO BETWEEN '{start}'
+                        AND '{end}'
+                        OR DT_FIM_VALIDADE BETWEEN '{start}'
+                        AND '{end}'
+                """,
+            },
+            "primary_key": [
+                "CD_CONSORCIO",
+                "CD_LINHA",
+            ],  # id column to nest data on
+            "interval_minutes": BILHETAGEM_TRATAMENTO_INTERVAL,
+        },
     ]
 
     BILHETAGEM_EXCLUDE = "+operadoras +consorcios +servicos"
@@ -470,15 +493,15 @@ class constants(Enum):  # pylint: disable=c0103
 aux_retorno_ordem_pagamento",
     }
 
-    BILHETAGEM_MATERIALIZACAO_TRANSACAO_RIOCARD_PARAMS = {
+    BILHETAGEM_MATERIALIZACAO_DASHBOARD_CONTROLE_VINCULO_PARAMS = {
         "dataset_id": "dashboard_controle_vinculo_jae_riocard",
         "table_id": "veiculo_indicadores_dia",
         "upstream": True,
         "dbt_vars": {
             "run_date": {},
             "version": {},
         },
-        "exclude": "+gps_sppo +sppo_licenciamento +gps_validador",
+        "exclude": "+gps_sppo +sppo_licenciamento +gps_validador +transacao_riocard",
     }
 
     BILHETAGEM_MATERIALIZACAO_ORDEM_PAGAMENTO_PARAMS = {

pipelines/migration/br_rj_riodejaneiro_bilhetagem/flows.py

@@ -2,7 +2,7 @@
 """
 Flows for br_rj_riodejaneiro_bilhetagem
 
-DBT: 2024-07-18
+DBT: 2024-08-05
 """
 
 from copy import deepcopy
@@ -217,28 +217,30 @@
 )
 
 
-bilhetagem_materializacao_transacao_riocard = deepcopy(default_materialization_flow)
-bilhetagem_materializacao_transacao_riocard.name = (
-    "SMTR: Bilhetagem Transação RioCard - Materialização"
+bilhetagem_materializacao_dashboard_controle_vinculo = deepcopy(default_materialization_flow)
+bilhetagem_materializacao_dashboard_controle_vinculo.name = (
+    "SMTR: Bilhetagem Controle Vinculo Validador - Materialização"
 )
-bilhetagem_materializacao_transacao_riocard.storage = GCS(smtr_constants.GCS_FLOWS_BUCKET.value)
-bilhetagem_materializacao_transacao_riocard.run_config = KubernetesRun(
+bilhetagem_materializacao_dashboard_controle_vinculo.storage = GCS(
+    smtr_constants.GCS_FLOWS_BUCKET.value
+)
+bilhetagem_materializacao_dashboard_controle_vinculo.run_config = KubernetesRun(
     image=smtr_constants.DOCKER_IMAGE.value,
     labels=[smtr_constants.RJ_SMTR_AGENT_LABEL.value],
 )
 
-bilhetagem_materializacao_transacao_riocard.state_handlers = [
+bilhetagem_materializacao_dashboard_controle_vinculo.state_handlers = [
     handler_inject_bd_credentials,
     handler_initialize_sentry,
     handler_skip_if_running,
 ]
 
-bilhetagem_materializacao_transacao_riocard = set_default_parameters(
-    flow=bilhetagem_materializacao_transacao_riocard,
-    default_parameters=constants.BILHETAGEM_MATERIALIZACAO_TRANSACAO_RIOCARD_PARAMS.value,
+bilhetagem_materializacao_dashboard_controle_vinculo = set_default_parameters(
+    flow=bilhetagem_materializacao_dashboard_controle_vinculo,
+    default_parameters=constants.BILHETAGEM_MATERIALIZACAO_DASHBOARD_CONTROLE_VINCULO_PARAMS.value,
 )
 
-bilhetagem_materializacao_transacao_riocard.schedule = every_day_hour_five
+bilhetagem_materializacao_dashboard_controle_vinculo.schedule = every_day_hour_five
 
 # Ordem Pagamento
 

queries/models/br_rj_riodejaneiro_bilhetagem/CHANGELOG.md

@@ -1,5 +1,19 @@
 # Changelog - bilhetagem
 
+## [2.2.0] - 2024-08-05
+
+### Adicionado
+- Cria modelo `aux_h3_res9.sql` para tratar os dados geográficos de tile (https://github.com/prefeitura-rio/pipelines_rj_smtr/pull/130)
+- Cria modelo `staging_linha_consorcio.sql` (https://github.com/prefeitura-rio/pipelines_rj_smtr/pull/130)
+
+### Alterado
+- `aux_passageiros_hora.sql` (https://github.com/prefeitura-rio/pipelines_rj_smtr/pull/130)
+  - Faz conversão dos dados geograficos no modelo
+  - Altera materialized de `ephemeral` para `table`
+  - Adiciona transações do tipo RioCard
+- Muda tipo das colunas de valor do modelo `staging_transacao_riocard.sql` de `float` para `numeric` (https://github.com/prefeitura-rio/pipelines_rj_smtr/pull/130)
+- Altera lógica incremental do modelo `transacao_riocard.sql` para rodar de hora em hora (https://github.com/prefeitura-rio/pipelines_rj_smtr/pull/130)
+
 ## [2.1.4] - 2024-08-02
 
 ### Alterado

queries/models/br_rj_riodejaneiro_bilhetagem/passageiros_hora.sql

@@ -11,11 +11,11 @@
 }}
 
 SELECT
-  * EXCEPT(id_transacao, latitude, longitude),
+  * EXCEPT(id_transacao, geo_point_transacao),
   COUNT(id_transacao) AS quantidade_passageiros,
   '{{ var("version") }}' AS versao
 FROM
-  {{ ref("passageiros_hora_aux") }}
+  {{ ref("aux_passageiros_hora") }}
 GROUP BY
   data,
   hora,

queries/models/br_rj_riodejaneiro_bilhetagem/passageiros_tile_hora.sql

@@ -11,16 +11,16 @@
 }}
 
 SELECT
-  p.* EXCEPT(id_transacao, latitude, longitude),
+  p.* EXCEPT(id_transacao, geo_point_transacao),
   geo.tile_id,
   COUNT(id_transacao) AS quantidade_passageiros,
   '{{ var("version") }}' AS versao
 FROM
-  {{ ref("passageiros_hora_aux") }} p
+  {{ ref("aux_passageiros_hora") }} p
 JOIN
-  {{ source("br_rj_riodejaneiro_geo", "h3_res9") }} geo
+  {{ ref("aux_h3_res9") }} geo
 ON
-  ST_CONTAINS(ST_GEOGFROMTEXT(geo.geometry), ST_GEOGPOINT(p.longitude, p.latitude))
+  ST_CONTAINS(geo.geometry, geo_point_transacao)
 GROUP BY
   data,
   hora,

queries/models/br_rj_riodejaneiro_bilhetagem/schema.yml

@@ -270,7 +270,7 @@ models:
       - name: versao
         description: "Código de controle de versão do dado (SHA Github)"
   - name: passageiros_tile_hora
-    description: "Tabela de contagem do número de passageiros por hora. Agrega valores da tabela de transações por: data, hora, modo, consorcio, operadora, servico, sentido, tipo_transacao e tile_id"
+    description: "Tabela de contagem do número de passageiros por hora. Agrega valores das tabelas transacao e transacao_riocard por: data, hora, modo, consorcio, operadora, servico, sentido, tipo_transacao e tile_id"
     columns:
       - name: data
         description: "Data da transação"
@@ -526,18 +526,30 @@ models:
         description: "Data e hora de processamento da transação em GMT-3 (formato YYYY-MM-ddTHH:mm:ss.ssssss)"
       - name: datetime_captura
         description: "Timestamp de captura em GMT-3 (formato YYYY-MM-dd HH:mm:ssTZD)"
-      - name: id_consorcio_jae
-        description: "Código do consórcio no sistema da Jaé"
-      - name: id_operadora_jae
-        description: "Código do operador no sistema da Jaé"
+      - name: modo
+        description: "Tipo de transporte (Ônibus, Van, BRT)"
+      - name: id_consorcio
+        description: "Identificador do consórcio na tabela cadastro.consorcios"
+      - name: consorcio
+        description: "Nome do consórcio"
+      - name: id_operadora
+        description: "Identificador da operadora na tabela cadastro.operadoras"
+      - name: operadora
+        description: "Nome da operadora de transporte (mascarado se for pessoa física)"
+      # - name: servico
+      #   description: "Nome curto da linha operada pelo veículo com variação de serviço (ex: 010, 011SN, ...) ou código da estação"
       - name: id_servico_jae
         description: "Identificador da linha no banco de dados da jaé (É possível cruzar os dados com a tabela rj-smtr.cadastro.servicos usando a coluna id_servico_jae)"
-      - name: id_validador
-        description: "Número de série do validador"
+      - name: servico_jae
+        description: "Nome curto da linha operada pelo veículo com variação de serviço (ex: 010, 011SN, ...) ou código da estação de BRT na Jaé"
+      - name: descricao_servico_jae
+        description: "Nome longo da linha operada pelo veículo com variação de serviço (ex: 010, 011SN, ...) ou nome da estação de BRT na Jaé"
       - name: sentido
         description: "Sentido de operação do serviço (0 = ida, 1 = volta)"
-      - name: id_cliente
-        description: "Identificador único do cliente"
+      - name: id_veiculo
+        description: "Identificador único do veículo"
+      - name: id_validador
+        description: "Número de série do validador"
       - name: id_transacao
         description: "Identificador único da transação"
       - name: latitude

queries/models/br_rj_riodejaneiro_bilhetagem/transacao.sql

@@ -136,9 +136,9 @@ new_data AS (
     l.nm_linha AS descricao_servico_jae,
     sentido,
     CASE
-    WHEN m.modo = "VLT" THEN SUBSTRING(t.veiculo_id, 1, 3)
-    WHEN m.modo = "BRT" THEN NULL
-    ELSE t.veiculo_id
+      WHEN m.modo = "VLT" THEN SUBSTRING(t.veiculo_id, 1, 3)
+      WHEN m.modo = "BRT" THEN NULL
+      ELSE t.veiculo_id
     END AS id_veiculo,
     t.numero_serie_validador AS id_validador,
     COALESCE(t.id_cliente, t.pan_hash) AS id_cliente,