GPS Viagens 2.0

queries/dbt_project.yml

@@ -53,6 +53,8 @@ vars:
   velocidade_limiar_parado: 3
   ## Distância mínima para que o veículo seja identificado parado em um terminal ou garagem em aux_registros_parada
   distancia_limiar_parada: 250
+  ## Tamanho do buffer do segmento
+  buffer_segmento_metros: 20 # flag_trajeto_correto
 
   ### [ANTIGO] SIGMOB ###
 
@@ -187,6 +189,9 @@ vars:
   ### Encontro de Contas ###
   encontro_contas_modo: ""
 
+  modo_gps: ""
+  fonte_gps: ""
+
 tests:
   rj_smtr:
     where: "DATA BETWEEN DATE('__date_range_start__') AND DATE('__date_range_end__')"

queries/models/monitoramento/gps.sql

@@ -0,0 +1,101 @@
+{{
+    config(
+        materialized="incremental",
+        partition_by={"field": "data", "data_type": "date", "granularity": "day"},
+        tags=["geolocalizacao"],
+        alias=this.name ~ var("modo_gps") ~ var("fonte_gps"),
+    )
+}}
+
+with
+    registros as (
+        -- 1. registros_filtrada
+        select
+            id_veiculo,
+            datetime_gps,
+            datetime_captura,
+            velocidade,
+            servico,
+            latitude,
+            longitude,
+        from {{ ref("aux_gps_filtrada") }}
+        {% if is_incremental() -%}
+            where
+                data between date("{{var('date_range_start')}}") and date(
+                    "{{var('date_range_end')}}"
+                )
+                and datetime_gps > "{{var('date_range_start')}}"
+                and datetime_gps <= "{{var('date_range_end')}}"
+        {%- endif -%}
+    ),
+    velocidades as (
+        -- 2. velocidades
+        select
+            id_veiculo,
+            datetime_gps,
+            servico,
+            velocidade,
+            distancia,
+            indicador_em_movimento
+        from {{ ref("aux_gps_velocidade") }}
+    ),
+    paradas as (
+        -- 3. paradas
+        select id_veiculo, datetime_gps, servico, tipo_parada,
+        from {{ ref("aux_gps_parada") }}
+    ),
+    indicadores as (
+        -- 4. indicador_trajeto_correto
+        select id_veiculo, datetime_gps, servico, route_id, indicador_trajeto_correto
+        from {{ ref("aux_gps_trajeto_correto") }}
+    )
+-- 5. Junção final
+select
+    date(r.datetime_gps) data,
+    r.datetime_gps,
+    r.id_veiculo,
+    r.servico,
+    r.latitude,
+    r.longitude,
+    case
+        when indicador_em_movimento is true and indicador_trajeto_correto is true
+        then 'Em operação'
+        when indicador_em_movimento is true and indicador_trajeto_correto is false
+        then 'Operando fora trajeto'
+        when indicador_em_movimento is false
+        then
+            case
+                when tipo_parada is not null
+                then concat("Parado ", tipo_parada)
+                else
+                    case
+                        when indicador_trajeto_correto is true
+                        then 'Parado trajeto correto'
+                        else 'Parado fora trajeto'
+                    end
+            end
+    end status,
+    r.velocidade as velocidade_instantanea,
+    v.velocidade as velocidade_estimada_10_min,
+    v.distancia,
+    '{{ var("version") }}' as versao,
+    current_datetime("America/Sao_Paulo") as datetime_ultima_atualizacao
+from registros r
+
+join
+    indicadores i
+using(id_veiculo, datetime_gps, servico)
+join
+    velocidades v
+using(id_veiculo, datetime_gps, servico)
+join
+    paradas p
+using(id_veiculo, datetime_gps, servico)
+{% if is_incremental() -%}
+    where
+        date(r.datetime_gps) between date("{{var('date_range_start')}}") and date(
+            "{{var('date_range_end')}}"
+        )
+        and r.datetime_gps > "{{var('date_range_start')}}"
+        and r.datetime_gps <= "{{var('date_range_end')}}"
+{%- endif -%}

queries/models/monitoramento/staging/aux_gps.sql

@@ -0,0 +1,19 @@
+{{ config(alias=this.name ~ var("fonte_gps")) }}
+
+select
+    data,
+    safe_cast(
+        datetime(timestamp(datahora), "America/Sao_Paulo") as datetime
+    ) datetime_gps,
+    safe_cast(ordem as string) id_veiculo,
+    concat(
+        ifnull(regexp_extract(linha, r'[A-Z]+'), ""),
+        ifnull(regexp_extract(linha, r'[0-9]+'), "")
+    ) as servico,
+    safe_cast(replace(latitude, ',', '.') as float64) latitude,
+    safe_cast(replace(longitude, ',', '.') as float64) longitude,
+    safe_cast(
+        datetime(timestamp(timestamp_captura), "America/Sao_Paulo") as datetime
+    ) datetime_captura,
+    safe_cast(velocidade as int64) velocidade
+from {{ source('br_rj_riodejaneiro_onibus_gps_' ~ var("fonte_gps") ~ '_staging', "registros") }}

queries/models/monitoramento/staging/aux_gps_filtrada.sql

@@ -0,0 +1,53 @@
+{{ config(materialized="ephemeral") }}
+
+with
+    box as (
+        /* 1. Geometria de caixa que contém a área do município de Rio de Janeiro.*/
+        select * from {{ var("limites_caixa") }}
+    ),
+    gps as (
+        /* 2. Filtra registros antigos. Remove registros que tem diferença maior que 1 minuto entre o timestamp_captura e datetime_gps.*/
+        select *, st_geogpoint(longitude, latitude) posicao_veiculo_geo
+        from {{ ref("aux_gps" ~ var("fonte_gps")) }}
+        where
+            data between date("{{var('date_range_start')}}") and date(
+                "{{var('date_range_end')}}"
+            )
+            {% if is_incremental() -%}
+                and datetime_gps > "{{var('date_range_start')}}"
+                and datetime_gps <= "{{var('date_range_end')}}"
+            {%- endif -%}
+    ),
+    realocacao as (
+        select g.* except (servico), coalesce(r.servico_realocado, g.servico) as servico
+        from gps g
+        left join
+            {{ ref("aux_gps_realocacao" ~ var("fonte_gps")) }} r
+            using(id_veiculo, datetime_gps)
+    ),
+    filtrada as (
+        /* 1,2, e 3. Muda o nome de variáveis para o padrão do projeto.*/
+        select
+            data,
+            datetime_gps,
+            id_veiculo,
+            servico,
+            latitude,
+            longitude,
+            posicao_veiculo_geo,
+            datetime_captura,
+            velocidade,
+            row_number() over (partition by id_veiculo, datetime_gps, servico) rn
+        from realocacao
+        where
+            st_intersectsbox(
+                posicao_veiculo_geo,
+                (select min_longitude from box),
+                (select min_latitude from box),
+                (select max_longitude from box),
+                (select max_latitude from box)
+            )
+    )
+select * except (rn), "{{ var('version') }}" as versao
+from filtrada
+where rn = 1

queries/models/monitoramento/staging/aux_gps_parada.sql

@@ -0,0 +1,101 @@
+{{ config(materialized="ephemeral") }}
+
+{% if execute %}
+    {% set feed_start_date = (
+        run_query(
+            "SELECT DISTINCT feed_start_date FROM rj-smtr-dev.rafael__planejamento.calendario WHERE data BETWEEN DATE('"
+            ~ var("date_range_start")
+            ~ "') AND DATE('"
+            ~ var("date_range_end")
+            ~ "')"
+        )
+        .columns[0]
+        .values()[0]
+    ) %}
+{% endif %}
+
+with
+    terminais as (
+        -- 1. Selecionamos terminais, criando uma geometria de ponto para cada.
+        select
+            st_geogpoint(stop_lon, stop_lat) as ponto_parada,
+            stop_name as nome_parada,
+            'terminal' as tipo_parada
+        -- from {{ ref("stops_gtfs") }}
+        from `rj-smtr`.`gtfs`.`stops`
+        where location_type = "1" and feed_start_date = date("{{ feed_start_date }}")
+    ),
+    garagens as (
+        -- 1. Selecionamos as garagens, , criando uma geometria de ponto para cada.
+        select distinct
+            st_astext(st_geogpoint(stop_lon, stop_lat)) as ponto_parada,
+            stop_name as nome_parada,
+            'garagens' as tipo_parada
+        -- from {{ ref("stops_gtfs") }}
+        from `rj-smtr`.`gtfs`.`stops`
+        left join
+            -- {{ ref("stop_times_gtfs") }} using (feed_version, feed_start_date,
+            -- stop_id)
+            `rj-smtr`.`gtfs`.`stop_times` using (feed_version, feed_start_date, stop_id)
+        where
+            pickup_type is null
+            and drop_off_type is null
+            and stop_name like "%Garagem%"
+            and feed_start_date = date("{{ feed_start_date }}")
+    ),
+    pontos_parada as (
+        -- Unimos terminais e garagens para obter todos os pontos de parada
+        select *
+        from terminais
+        union all
+        select  st_geogfromtext(ponto_parada) as ponto_parada,
+                * except(ponto_parada)
+        from garagens
+    ),
+    distancia as (
+        -- 2. Calculamos as distâncias e definimos nrow
+        select
+            id_veiculo,
+            datetime_gps,
+            data,
+            servico,
+            posicao_veiculo_geo,
+            nome_parada,
+            tipo_parada,
+            round(st_distance(posicao_veiculo_geo, ponto_parada), 1) distancia_parada,
+            row_number() over (
+                partition by datetime_gps, id_veiculo, servico
+                order by st_distance(posicao_veiculo_geo, ponto_parada)
+            ) nrow
+        from pontos_parada p
+        join
+            (
+                select id_veiculo, datetime_gps, data, servico, posicao_veiculo_geo
+                from {{ ref("aux_gps_filtrada") }}
+                {% if not flags.FULL_REFRESH %}
+                    where
+                        data between date("{{var('date_range_start')}}") and date(
+                            "{{var('date_range_end')}}"
+                        )
+                        and datetime_gps > "{{var('date_range_start')}}"
+                        and datetime_gps <= "{{var('date_range_end')}}"
+                {% endif %}
+            ) r
+            on 1 = 1
+    )
+select
+    data,
+    datetime_gps,
+    id_veiculo,
+    servico,
+    /*
+  3. e 4. Identificamos o status do veículo como 'terminal', 'garagem' (para os veículos parados) ou
+  null (para os veículos mais distantes de uma parada que o limiar definido)
+  */
+    case
+        when distancia_parada < {{ var("distancia_limiar_parada") }}
+        then tipo_parada
+        else null
+    end tipo_parada,
+from distancia
+where nrow = 1

queries/models/monitoramento/staging/aux_gps_realocacao.sql

@@ -0,0 +1,70 @@
+{% if var("fifteen_minutes") == "_15_minutos" %}
+    {{ config(materialized="ephemeral", alias=this.name ~ var("fonte_gps")) }}  -- verificar isso
+{% else %}
+    {{
+        config(
+            materialized="incremental",
+            partition_by={"field": "data", "data_type": "date", "granularity": "day"},
+            alias=this.name ~ var("fonte_gps"),
+        )
+    }}
+{% endif %}
+
+-- 1. Filtra realocações válidas dentro do intervalo de GPS avaliado
+with
+    realocacao as (
+        select
+            * except (datetime_saida),
+            case
+                when datetime_saida is null then datetime_operacao else datetime_saida
+            end as datetime_saida,
+        from {{ ref("aux_realocacao" ~ var("fonte_gps")) }}
+        where
+            -- Realocação deve acontecer após o registro de GPS e até 1 hora depois
+            datetime_diff(datetime_operacao, datetime_entrada, minute) between 0 and 60
+            and data between date("{{var('date_range_start')}}") and date(
+                datetime_add("{{var('date_range_end')}}", interval 1 hour)
+            )
+            and (
+                datetime_saida >= datetime("{{var('date_range_start')}}")
+                or datetime_operacao >= datetime("{{var('date_range_start')}}")
+            )
+    ),
+    -- 2. Altera registros de GPS com servicos realocados
+    gps as (
+        select id_veiculo, datetime_gps, servico, data,
+        from {{ ref("aux_gps" ~ var("fonte_gps")) }}
+        where
+            data between date("{{var('date_range_start')}}") and date(
+                "{{var('date_range_end')}}"
+            )
+            and datetime_gps > "{{var('date_range_start')}}"
+            and datetime_gps <= "{{var('date_range_end')}}"
+    ),
+    combinacao as (
+        select
+            g.data,
+            g.datetime_gps,
+            r.id_veiculo,
+            g.servico as servico_gps,
+            r.servico as servico_realocado,
+            r.datetime_operacao as datetime_realocado
+        from gps g
+        inner join
+            realocacao r
+            on g.id_veiculo = r.id_veiculo
+            and g.servico != r.servico
+            and g.datetime_gps between r.datetime_entrada and r.datetime_saida
+    )
+-- Filtra realocacao mais recente para cada timestamp
+select * except (rn)
+from
+    (
+        select
+            *,
+            row_number() over (
+                partition by id_veiculo, datetime_gps order by datetime_realocado desc
+            ) as rn
+        from combinacao
+    )
+where rn = 1