Refatorar dataset veiculos

pipelines/migration/veiculo/CHANGELOG.md

@@ -1,5 +1,12 @@
 # Changelog - veiculo
 
+## [1.1.0] - 2024-08-06
+
+### Alterado
+- Alterações no tratamento do arquivo de infrações (https://github.com/prefeitura-rio/pipelines_rj_smtr/pull/126):
+  - Remove coluna `placa` das primary keys
+  - Remove filtro de modo
+
 ## [1.0.1] - 2024-05-28
 
 ### Adicionado

pipelines/migration/veiculo/constants.py

@@ -13,7 +13,7 @@ class constants(Enum):  # pylint: disable=c0103
     Constant values for rj_smtr veiculo
     """
 
-    SPPO_LICENCIAMENTO_TABLE_ID = "sppo_licenciamento_stu"
+    SPPO_LICENCIAMENTO_TABLE_ID = "licenciamento_stu"
 
     SPPO_LICENCIAMENTO_MAPPING_KEYS = {
         "placa": "placa",
@@ -46,7 +46,7 @@ class constants(Enum):  # pylint: disable=c0103
         "names": SPPO_LICENCIAMENTO_MAPPING_KEYS.keys(),  # pylint: disable=e1101
     }
 
-    SPPO_INFRACAO_TABLE_ID = "sppo_infracao"
+    SPPO_INFRACAO_TABLE_ID = "infracao"
 
     SPPO_INFRACAO_MAPPING_KEYS = {
         "permissao": "permissao",

pipelines/migration/veiculo/tasks.py

@@ -13,7 +13,7 @@
 from prefect import task
 
 from pipelines.constants import constants as smtr_constants
-from pipelines.migration.utils import connect_ftp, data_info_str, filter_data
+from pipelines.migration.utils import connect_ftp, data_info_str
 from pipelines.migration.veiculo.constants import constants
 from pipelines.utils.utils import log  # ,get_vault_secret
 
@@ -200,13 +200,13 @@ def pre_treatment_sppo_infracao(status: dict, timestamp: datetime):
         log("Updating valor type to float...", level="info")
         data["valor"] = data["valor"].str.replace(",", ".").astype(float)
 
-        filters = ["modo != 'ONIBUS'"]
-        log(f"Filtering '{filters}'...", level="info")
-        data = filter_data(data, filters)
+        # filters = ["modo != 'ONIBUS'"]
+        # log(f"Filtering '{filters}'...", level="info")
+        # data = filter_data(data, filters)
 
-        log("Filtering null primary keys...", level="info")
-        primary_key = ["placa", "id_auto_infracao"]
-        data.dropna(subset=primary_key, inplace=True)
+        # log("Filtering null primary keys...", level="info")
+        primary_key = ["id_auto_infracao"]
+        # data.dropna(subset=primary_key, inplace=True)
 
         # Check primary keys
         # pk_columns = ["placa", "id_auto_infracao"]

queries/macros/get_license_date.sql

@@ -10,7 +10,7 @@ SELECT
     WHEN DATE("{{ var('run_date') }}") >= "2024-03-16" AND DATE("{{ var('run_date') }}") < "2024-04-01" THEN DATE("2024-04-09")
     ELSE (
       SELECT MIN(DATE(data))
-      FROM {{ ref("sppo_licenciamento_stu_staging") }}
+      FROM {{ ref("licenciamento_stu_staging") }}
       WHERE DATE(data) >= DATE_ADD(DATE("{{ var('run_date') }}"), INTERVAL 5 DAY)
         -- Admite apenas versões do STU igual ou após 2024-04-09 a partir de abril/24 devido à falha de atualização na fonte da dados (SIURB)
         AND (DATE("{{ var('run_date') }}") < "2024-04-01" OR DATE(data) >= '2024-04-09')

queries/models/sources.yml

@@ -110,10 +110,10 @@ sources:
       - name: sppo_vistoria_tr_subtt_cglf_2024
       - name: sppo_vistoria_tr_subtt_cglf_pendentes_2024
       - name: sppo_registro_agente_verao
-      - name: sppo_infracao
+      - name: infracao
       - name: sppo_licenciamento_solicitacao
       - name: sppo_licenciamento
-      - name: sppo_licenciamento_stu
+      - name: licenciamento_stu
 
   - name: dashboard_subsidio_sppo_staging
     database: rj-smtr-staging

queries/models/veiculo/CHANGELOG.md

@@ -1,5 +1,14 @@
 # Changelog - veiculo
 
+## [2.0.0] - 2024-08-06
+
+### Adicionado
+- Criado os modelos `infracao.sql` e `licenciamento.sql` para guardar os dados completos de infração e licenciamento do STU (https://github.com/prefeitura-rio/pipelines_rj_smtr/pull/126)
+
+### Alterado
+- Altera tipo dos modelos `sppo_infracao.sql` e `sppo_licenciamento.sql` para `ephemeral` (https://github.com/prefeitura-rio/pipelines_rj_smtr/pull/126)
+- Adiciona colunas `placa`, `data_licenciamento`, `data_infracao` e `datetime_ultima_atualizacao` no modelo `sppo_veiculo_dia.sql` (https://github.com/prefeitura-rio/pipelines_rj_smtr/pull/126)
+
 ## [1.1.3] - 2024-10-25
 
 #### Alterado
@@ -12,36 +21,36 @@
 
 ## [1.1.2] - 2024-04-25
 
-#### Adicionado
+### Adicionado
 
 - Criada macro `get_license_date.sql` para retornar relacionamento entre `run_date` e data de licenciamento dos modelos `sppo_licenciamento_stu_staging.sql`, `sppo_licenciamento.sql` e `sppo_veiculo_dia.sql`. Nesta macro, serão admitidas apenas versões do STU igual ou após 2024-04-09 a partir de abril/24 devido à falha de atualização na fonte da dados (SIURB) (https://github.com/prefeitura-rio/queries-rj-smtr/pull/280)
 
-#### Corrigido
+### Corrigido
 
 - Altera lógica de particionamento nos modelos `sppo_licenciamento_stu_staging.sql`, `sppo_licenciamento.sql`, `sppo_infracao_staging.sql` e `sppo_infracao.sql` (https://github.com/prefeitura-rio/queries-rj-smtr/pull/280)
 
 ## [1.1.1] - 2024-04-16
 
-#### Corrigido
+### Corrigido
 
 - Cria lógica de deduplicação na tabela `sppo_registro_agente_verao` (https://github.com/prefeitura-rio/queries-rj-smtr/pull/271)
 
 ## [1.1.0] - 2024-04-15
 
-#### Alterado
+### Alterado
 
 - Reorganizar modelos intermediários de licenciamento em staging (https://github.com/prefeitura-rio/queries-rj-smtr/pull/255)
 - Atualiza schema para refletir as alterações (https://github.com/prefeitura-rio/queries-rj-smtr/pull/255)
 
 ## [1.0.2] - 2024-04-12
 
-#### Alterado
+### Alterado
 
 - Fixa versão do STU em `2024-04-09` para mar/Q2 devido à falha de atualização na fonte da dados (SIURB) nos modelos `sppo_licenciamento.sql` e `sppo_veiculo_dia.sql` (https://github.com/prefeitura-rio/queries-rj-smtr/pull/264)
 
 ## [1.0.1] - 2024-04-05
 
-#### Alterado
+### Alterado
 - Altera a localização da verificação de validade da vistoria de `sppo_licenciamento` para `sppo_veiculo_dia` (https://github.com/prefeitura-rio/queries-rj-smtr/pull/252)
 - Adiciona coluna `data_inicio_veiculo` na tabela `sppo_licenciamento` (https://github.com/prefeitura-rio/queries-rj-smtr/pull/252)
 
@@ -65,7 +74,7 @@
 - Atualiza documentação de tabelas e colunas (https://github.com/prefeitura-rio/queries-rj-smtr/pull/239)
 - Alterações feitas em https://github.com/prefeitura-rio/queries-rj-smtr/pull/229 e https://github.com/prefeitura-rio/queries-rj-smtr/pull/236 corrigidas em https://github.com/prefeitura-rio/queries-rj-smtr/pull/239
 
-## Corrigido
+### Corrigido
 
 - Corrige versão dos dados de licenciamento do STU a partir de 01/03/24
   na tabela `sppo_licenciamento` (https://github.com/prefeitura-rio/queries-rj-smtr/pull/239)

queries/models/veiculo/infracao.sql

@@ -0,0 +1,44 @@
+
+{{
+  config(
+    materialized='incremental',
+    partition_by={
+      "field":"data",
+      "data_type": "date",
+      "granularity":"day"
+    },
+    unique_key=['data', 'id_auto_infracao'],
+    incremental_strategy='insert_overwrite'
+  )
+}}
+
+{%- if execute and is_incremental() %}
+  {% set infracao_date = run_query("SELECT MIN(SAFE_CAST(data AS DATE)) FROM " ~ ref('infracao_staging') ~ " WHERE SAFE_CAST(data AS DATE) >= DATE_ADD(DATE('" ~ var("run_date") ~ "'), INTERVAL 7 DAY)").columns[0].values()[0] %}
+{% endif -%}
+
+WITH infracao AS (
+  SELECT
+    * EXCEPT(data),
+    SAFE_CAST(data AS DATE) AS data
+  FROM
+    {{ ref("infracao_staging") }} as t
+  {% if is_incremental() %}
+    WHERE
+      DATE(data) = DATE("{{ infracao_date }}")
+  {% endif %}
+),
+infracao_rn AS (
+  SELECT
+    *,
+    ROW_NUMBER() OVER (PARTITION BY data, id_auto_infracao ORDER BY timestamp_captura DESC) rn
+  FROM
+    infracao
+)
+SELECT
+  * EXCEPT(rn),
+  CURRENT_DATETIME("America/Sao_Paulo") AS datetime_ultima_atualizacao,
+  "{{ var("version") }}" AS versao
+FROM
+  infracao_rn
+WHERE
+  rn = 1

queries/models/veiculo/licenciamento.sql

@@ -0,0 +1,88 @@
+-- depends_on: {{ ref('aux_sppo_licenciamento_vistoria_atualizada') }}
+{{
+  config(
+    materialized="incremental",
+    partition_by={"field": "data", "data_type": "date", "granularity": "day"},
+    unique_key=["data", "id_veiculo"],
+    incremental_strategy="insert_overwrite",
+  )
+}}
+
+{% if execute and is_incremental() %}
+  {% set licenciamento_date = run_query(get_license_date()).columns[0].values()[0] %}
+{% endif %}
+
+WITH stu AS (
+  SELECT
+    * EXCEPT(data),
+    DATE(data) AS data
+  FROM
+    {{ ref("licenciamento_stu_staging") }} AS t
+  {% if is_incremental() %}
+    WHERE
+      DATE(data) = DATE("{{ licenciamento_date }}")
+  {% endif %}
+),
+stu_rn AS (
+  SELECT
+    * EXCEPT (timestamp_captura),
+    EXTRACT(YEAR FROM data_ultima_vistoria) AS ano_ultima_vistoria,
+    ROW_NUMBER() OVER (PARTITION BY data, id_veiculo) rn
+  FROM
+    stu
+),
+stu_ano_ultima_vistoria AS (
+  -- Temporariamente considerando os dados de vistoria enviados pela TR/SUBTT/CGLF
+  SELECT
+    s.* EXCEPT(ano_ultima_vistoria),
+    CASE
+      WHEN data >= "2024-03-01" AND c.ano_ultima_vistoria > s.ano_ultima_vistoria THEN c.ano_ultima_vistoria
+      WHEN data >= "2024-03-01" THEN COALESCE(s.ano_ultima_vistoria, c.ano_ultima_vistoria)
+      ELSE s.ano_ultima_vistoria
+    END AS ano_ultima_vistoria_atualizado,
+  FROM
+    stu_rn AS s
+  LEFT JOIN
+  (
+    SELECT
+      id_veiculo,
+      placa,
+      ano_ultima_vistoria
+    FROM
+      {{ ref("aux_sppo_licenciamento_vistoria_atualizada") }}
+  ) AS c
+  USING(id_veiculo, placa)
+)
+SELECT
+  data,
+  modo,
+  id_veiculo,
+  ano_fabricacao,
+  carroceria,
+  data_ultima_vistoria,
+  id_carroceria,
+  id_chassi,
+  id_fabricante_chassi,
+  id_interno_carroceria,
+  id_planta,
+  indicador_ar_condicionado,
+  indicador_elevador,
+  indicador_usb,
+  indicador_wifi,
+  nome_chassi,
+  permissao,
+  placa,
+  quantidade_lotacao_pe,
+  quantidade_lotacao_sentado,
+  tipo_combustivel,
+  tipo_veiculo,
+  status,
+  data_inicio_vinculo,
+  ano_ultima_vistoria_atualizado,
+  CURRENT_DATETIME("America/Sao_Paulo") AS datetime_ultima_atualizacao,
+  "{{ var("version") }}" AS versao
+FROM
+  stu_ano_ultima_vistoria
+WHERE
+  rn = 1
+

queries/models/veiculo/schema.yml

@@ -1,7 +1,7 @@
 version: 2
 
 models:
-  - name: sppo_infracao
+  - name: infracao
     description: "Tabela histórica de todas as multas aplicadas aos modos de transporte no município do Rio de Janeiro, com qualquer situação.
     Dados capturados diariamente de: https://www.data.rio/documents/multas-aplicadas-aos-modos-de-transporte-nos-%C3%BAltimos-cinco-anos"
     columns:
@@ -34,7 +34,11 @@ models:
                       CANCELADA - Multa foi cancelada através de um Processo de Recurso"
       - name: data_pagamento
         description: "Data de pagamento"
-  - name: sppo_licenciamento
+      - name: datetime_ultima_atualizacao
+        description: "Última atualização (GMT-3)."
+      - name: versao
+        description: "SHA do repositório no Github"
+  - name: licenciamento
     description: "Tabela histórica de dados cadastrais dos veículos que operam o sistema de ônibus (BRT e SPPO),
                   considerando tanto os licenciados no Sistema de Transporte Urbano (STU) quanto as solicitações
                   válidas em andamento para ingresso no sistema"
@@ -90,6 +94,10 @@ models:
         description: "Ano atualizado da última vistoria realizada pelo veículo"
       - name: data_inicio_vinculo
         description: "Data de início do vínculo do veículo no STU"
+      - name: datetime_ultima_atualizacao
+        description: "Última atualização (GMT-3)."
+      - name: versao
+        description: "SHA do repositório no Github"
   - name: sppo_licenciamento_solicitacao
     description: "Tabela histórica de dados cadastrais das solicitações em andamento para ingresso no Sistema de Transporte Urbano (STU)"
     columns:
@@ -153,6 +161,8 @@ models:
         description: "Data de operação"
       - name: id_veiculo
         description: "Número de ordem do veículo, identificação que é exibida na lateral do ônibus"
+      - name: placa
+        description: "Placa do veículo"
       - name: indicadores
         description: "Indicadores para caraterização do status do veículo"
       - name: indicadores.indicador_licenciado
@@ -179,6 +189,12 @@ models:
                       - Autuado por limpeza/equipamento - Veículo que operou, foi licenciado, mas foi autuado cumulativamente por infrações relacionadas à limpeza e equipamentos do veículo\n
                       - Sem ar e não autuado - Veículo que operou, foi licenciado sem ar condicionado e não foi autuado\n
                       - Com ar e não autuado - Veículo que operou, foi licenciado com ar condicionado e não foi autuado"
+      - name: datetime_ultima_atualizacao
+        description: "Última atualização (GMT-3)."
+      - name: data_licenciamento
+        description: "Data do arquivo de licenciamento utilizado"
+      - name: data_infracao
+        description: "Data do arquivo de infrações utilizado"
       - name: versao
         description: "SHA do repositório no Github"
   - name: sppo_registro_agente_verao

queries/models/veiculo/sppo_infracao.sql

@@ -1,40 +1,13 @@
-
- {{ config(
-       materialized='incremental',
-       partition_by={
-              "field":"data",
-              "data_type": "date",
-              "granularity":"day"
-       },
-       unique_key=['data', 'id_auto_infracao'],
-       incremental_strategy='insert_overwrite'
-)
+{{
+  config(
+    materialized="ephemeral"
+  )
 }}
 
-{%- if execute %}
-  {% set infracao_date = run_query("SELECT MIN(SAFE_CAST(data AS DATE)) FROM " ~ ref('sppo_infracao_staging') ~ " WHERE SAFE_CAST(data AS DATE) >= DATE_ADD(DATE('" ~ var("run_date") ~ "'), INTERVAL 7 DAY)").columns[0].values()[0] %}
-{% endif -%}
-
-WITH
-  infracao AS (
-    SELECT
-      * EXCEPT(data),
-      SAFE_CAST(data AS DATE) AS data
-    FROM
-      {{ ref("sppo_infracao_staging") }} as t
-    WHERE
-      DATE(data) = DATE("{{ infracao_date }}")
-  ),
-  infracao_rn AS (
-    SELECT
-      *,
-      ROW_NUMBER() OVER (PARTITION BY data, id_auto_infracao) rn
-    FROM
-      infracao
-  )
 SELECT
-  * EXCEPT(rn)
+  *
 FROM
-  infracao_rn
+  {{ ref("infracao") }}
 WHERE
-  rn = 1
+  modo = 'ONIBUS'
+  AND placa IS NOT NULL