Merge branch 'staging/viagens-2-ajustes' of https://github.com/prefei…

…tura-rio/pipelines_rj_smtr into staging/viagens-2-ajustes

pipelines/migration/br_rj_riodejaneiro_gtfs/flows.py

@@ -2,7 +2,7 @@
 """
 Flows for gtfs
 
-DBT 2025-01-13
+DBT 2025-01-16
 """
 
 from prefect import Parameter, case, task

queries/dbt_project.yml

@@ -108,6 +108,7 @@ vars:
   shapes_version: "YYYY-MM-DD"
   frequencies_version: "YYYY-MM-DD"
   tipo_materializacao: "subsidio"
+  data_inicio_trips_shapes: "2023-04-01"
   # Feature penalidade de autuação por inoperância do ar condicionado (DECRETO RIO 51940/2023)
   DATA_SUBSIDIO_V2_INICIO: "2023-01-16"
   # Feature penalidade de autuação por segurança e limpeza/equipamento (DECRETO RIO 52820/2023)
@@ -377,3 +378,10 @@ models:
     datario:
       +materialized: view
       +schema: datario
+    dashboard_operacao_onibus:
+      +materialized: incremental
+      +incremental_strategy: insert_overwrite
+      +schema: dashboard_operacao_onibus
+      staging:
+        +materialized: view
+        +schema: dashboard_operacao_onibus_staging

queries/models/dashboard_operacao_onibus/CHANGELOG.md

@@ -0,0 +1,7 @@
+# Changelog - dashboard_operacao_onibus
+
+## [1.0.0] - 2025-01-16
+
+### Adicionado
+
+- Cria novo dataset para views utilizadas no dashboard de operação dos ônibus (https://github.com/prefeitura-rio/pipelines_rj_smtr/pull/389)

queries/models/dashboard_operacao_onibus/ordem_servico_diaria.sql

@@ -0,0 +1,24 @@
+{{
+    config(
+        materialized="view",
+        labels={
+            "dashboard": "yes",
+        },
+    )
+}}
+
+with
+    ordem_servico_diaria as (
+        select *
+        from {{ ref("aux_ordem_servico_diaria_v1") }}
+        where data < "{{ var('data_inicio_trips_shapes') }}"
+
+        union all
+
+        select *
+        from {{ ref("aux_ordem_servico_diaria_v2") }}
+        where data >= "{{ var('data_inicio_trips_shapes') }}"
+    )
+select *
+from ordem_servico_diaria
+where viagens_planejadas > 0

queries/models/dashboard_operacao_onibus/schema.yml

@@ -0,0 +1,24 @@
+version: 2
+
+models:
+  - name: ordem_servico_diaria
+    description: "Dados diários das linhas presentes no município."
+    columns:
+      - name: DATA
+        description: "Data da ordem de serviço."
+      - name: tipo_dia
+        description: "{{ doc('tipo_dia') }}"
+      - name: servico
+        description: "{{ doc('servico') }}"
+      - name: vista
+        description: "{{ doc('vista') }}"
+      - name: consorcio
+        description: "{{ doc('consorcio') }}"
+      - name: sentido
+        description: "{{ doc('sentido') }}"
+      - name: viagens_planejadas
+        description: "{{ doc('viagens_planejadas') }}"
+      - name: inicio_periodo
+        description: "{{ doc('inicio_periodo') }}"
+      - name: fim_periodo
+        description: "{{ doc('fim_periodo') }}"

queries/models/dashboard_operacao_onibus/staging/aux_ordem_servico_diaria_v1.sql

@@ -0,0 +1,85 @@
+{{ config(materialized="ephemeral") }}
+
+with
+    feed_start_date as (
+        select
+            feed_start_date,
+            feed_start_date as data_inicio,
+            coalesce(
+                date_sub(
+                    lead(feed_start_date) over (order by feed_start_date),
+                    interval 1 day
+                ),
+                last_day(feed_start_date, month)
+            ) as data_fim
+        from (select distinct feed_start_date from {{ ref("ordem_servico_gtfs") }})
+    ),
+    ordem_servico_pivot as (
+        select *
+        from
+            {{ ref("ordem_servico_gtfs") }} pivot (
+                max(partidas_ida) as partidas_ida,
+                max(partidas_volta) as partidas_volta,
+                max(viagens_planejadas) as viagens_planejadas,
+                max(distancia_total_planejada) as km for
+                tipo_dia in (
+                    'Dia Útil' as du,
+                    'Ponto Facultativo' as pf,
+                    'Sabado' as sab,
+                    'Domingo' as dom
+                )
+            )
+    ),
+    subsidio_feed_start_date_efetiva as (
+        select
+            data, split(tipo_dia, " - ")[0] as tipo_dia, tipo_dia as tipo_dia_original
+        from {{ ref("subsidio_data_versao_efetiva") }}
+    )
+select
+    data,
+    tipo_dia_original as tipo_dia,
+    servico,
+    vista,
+    consorcio,
+    sentido,
+    case
+        {% set tipo_dia = {
+            "Dia Útil": "du",
+            "Ponto Facultativo": "pf",
+            "Sabado": "sab",
+            "Domingo": "dom",
+        } %}
+        {% set sentido = {"ida": ("I", "C"), "volta": "V"} %}
+        {%- for key_s, value_s in sentido.items() %}
+            {%- for key_td, value_td in tipo_dia.items() %}
+                when
+                    sentido
+                    {% if key_s == "ida" %} in {{ value_s }}
+                    {% else %} = "{{ value_s }}"
+                    {% endif %} and tipo_dia = "{{ key_td }}"
+                then
+                    {% if key_td in ["Sabado", "Domingo"] %}
+                        round(
+                            safe_divide(
+                                (partidas_{{ key_s }}_du * km_{{ value_td }}), km_du
+                            )
+                        )
+                    {% else %} partidas_{{ key_s }}_{{ value_td }}
+                    {% endif %}
+            {% endfor -%}
+        {% endfor -%}
+    end as viagens_planejadas,
+    horario_inicio as inicio_periodo,
+    horario_fim as fim_periodo
+from
+    unnest(
+        generate_date_array(
+            (select min(data_inicio) from feed_start_date),
+            (select max(data_fim) from feed_start_date)
+        )
+    ) as data
+left join feed_start_date as d on data between d.data_inicio and d.data_fim
+left join subsidio_feed_start_date_efetiva as sd using (data)
+left join ordem_servico_pivot as o using (feed_start_date)
+left join {{ ref("servicos_sentido") }} using (feed_start_date, servico)
+where data < "{{ var('data_inicio_trips_shapes') }}"

queries/models/dashboard_operacao_onibus/staging/aux_ordem_servico_diaria_v2.sql

@@ -0,0 +1,104 @@
+{{ config(materialized="ephemeral") }}
+
+
+with
+    feed_info as (select * from `rj-smtr`.`gtfs`.`feed_info`),
+    ordem_servico_pivot as (
+        select *
+        from
+            {{ ref("ordem_servico_gtfs") }} pivot (
+                max(partidas_ida) as partidas_ida,
+                max(partidas_volta) as partidas_volta,
+                max(viagens_planejadas) as viagens_planejadas,
+                max(distancia_total_planejada) as km for tipo_dia in (
+                    'Dia Útil' as du,
+                    'Ponto Facultativo' as pf,
+                    'Sabado' as sab,
+                    'Domingo' as dom
+                )
+            )
+    ),
+    ordem_servico_regular as (
+        select feed_start_date, servico, partidas_ida_du, partidas_volta_du, km_du
+        from ordem_servico_pivot
+        where tipo_os = "Regular"
+    ),
+    ordem_servico_tratada as (
+        select
+
+            osp.* except (partidas_ida_du, partidas_volta_du, km_du),
+            case
+                when osp.partidas_ida_du = 0 or osp.partidas_ida_du is null
+                then osr.partidas_ida_du
+                else osp.partidas_ida_du
+            end as partidas_ida_du,
+            case
+                when osp.partidas_volta_du = 0 or osp.partidas_volta_du is null
+                then osr.partidas_volta_du
+                else osp.partidas_volta_du
+            end as partidas_volta_du,
+            case
+                when osp.km_du = 0 or osp.km_du is null then osr.km_du else osp.km_du
+            end as km_du
+        from ordem_servico_pivot osp
+        left join ordem_servico_regular osr using (feed_start_date, servico)
+    ),
+    subsidio_feed_start_date_efetiva as (
+        select
+            data,
+            tipo_os,
+            split(tipo_dia, " - ")[0] as tipo_dia,
+            tipo_dia as tipo_dia_original
+        from {{ ref("subsidio_data_versao_efetiva") }}
+    ),
+    ordem_servico_trips_shapes as (
+        select distinct feed_start_date, servico, tipo_os, sentido
+        from {{ ref("ordem_servico_trips_shapes_gtfs") }}
+    )
+select
+    data,
+    tipo_dia_original as tipo_dia,
+    servico,
+    vista,
+    consorcio,
+    sentido,
+    case
+        {% set tipo_dia = {
+            "Dia Útil": "du",
+            "Ponto Facultativo": "pf",
+            "Sabado": "sab",
+            "Domingo": "dom",
+        } %}
+        {% set sentido = {"ida": ("I", "C"), "volta": "V"} %}
+        {%- for key_s, value_s in sentido.items() %}
+            {%- for key_td, value_td in tipo_dia.items() %}
+                when
+                    sentido
+                    {% if key_s == "ida" %} in {{ value_s }}
+                    {% else %} = "{{ value_s }}"
+                    {% endif %} and tipo_dia = "{{ key_td }}"
+                then
+                    {% if key_td in ["Sabado", "Domingo"] %}
+                        round(
+                            safe_divide(
+                                (partidas_{{ key_s }}_du * km_{{ value_td }}), km_du
+                            )
+                        )
+                    {% else %} partidas_{{ key_s }}_{{ value_td }}
+                    {% endif %}
+            {% endfor -%}
+        {% endfor -%}
+    end as viagens_planejadas,
+    horario_inicio as inicio_periodo,
+    horario_fim as fim_periodo
+from
+    unnest(
+        generate_date_array(
+            (select min(feed_start_date) from feed_info),
+            (select max(feed_end_date) from feed_info)
+        )
+    ) as data
+left join feed_info as d on data between d.feed_start_date and d.feed_end_date
+left join ordem_servico_tratada as o using (feed_start_date)
+inner join subsidio_feed_start_date_efetiva as sd using (data, tipo_os)
+left join ordem_servico_trips_shapes using (feed_start_date, servico, tipo_os)

queries/models/dashboard_operacao_onibus/staging/servicos_sentido.sql

@@ -0,0 +1,67 @@
+{{ config(materialized="view") }}
+
+with
+    servicos_exclusivos_sabado as (
+        select distinct servico
+        from {{ ref("ordem_servico_gtfs") }}
+        where tipo_dia = "Dia Útil" and viagens_planejadas = 0
+    ),
+    servicos as (
+        select * except (versao_modelo, shape)
+        from {{ ref("trips_gtfs") }} as t
+        left join {{ ref("shapes_geom_gtfs") }} as s using (feed_start_date, shape_id)
+        where
+            (
+                feed_start_date >= "2023-06-01"
+                and (
+                    trip_short_name not in (select * from servicos_exclusivos_sabado)
+                    and (service_id like "U_R%" or service_id like "U_O%")
+                )
+                or (
+                    trip_short_name in (select * from servicos_exclusivos_sabado)
+                    and (service_id like "S_R%" or service_id like "S_O%")
+                )
+            )
+            or (
+                feed_start_date < "2023-06-01"
+                and (
+                    trip_short_name not in (select * from servicos_exclusivos_sabado)
+                    or (
+                        trip_short_name in (select * from servicos_exclusivos_sabado)
+                        and service_id = "S"
+                    )
+                )
+            )
+            and shape_distance is not null
+    ),
+    servicos_rn as (
+        select
+            *,
+            row_number() over (
+                partition by feed_start_date, trip_short_name, direction_id
+                order by trip_short_name, service_id, shape_id, direction_id
+            ) as rn
+        from servicos
+    ),
+    servicos_filtrada as (select * except (rn) from servicos_rn where rn = 1),
+    servicos_potencialmente_circulares as (
+        select
+            feed_start_date, trip_short_name, count(distinct direction_id) as q_direcoes
+        from servicos_filtrada
+        group by 1, 2
+        having count(distinct direction_id) = 1
+    )
+select
+    feed_start_date,
+    trip_short_name as servico,
+    case
+        when q_direcoes = 1 and st_distance(start_pt, end_pt) <= 50
+        then "C"
+        when direction_id = "0"
+        then "I"
+        when direction_id = "1"
+        then "V"
+    end as sentido
+from servicos_filtrada as sf
+left join
+    servicos_potencialmente_circulares as spc using (feed_start_date, trip_short_name)

queries/models/gtfs/CHANGELOG.md

@@ -1,5 +1,15 @@
 # Changelog - gtfs
 
+## [1.2.1] - 2025-01-16
+
+### Alterado
+
+- Move modelos `ordem_servico_diaria.sql` e `servicos_sentido.sql` para o dataset `dashboard_operacao_onibus` (https://github.com/prefeitura-rio/pipelines_rj_smtr/pull/389)
+
+### Removido
+
+- Modelo `ordem_servico_viagens_planejadas.sql` deletado
+
 ## [1.2.0] - 2024-12-04
 
 ### Alterado