add service alert trip summaries and fix typos

cal-itp · Jan 23, 2023 · 6879c31 · 6879c31
1 parent 114bd66
commit 6879c31
Show file tree

Hide file tree

Showing 4 changed files with 63 additions and 2 deletions.
diff --git a/warehouse/models/intermediate/gtfs/int_gtfs_rt__service_alerts_trip_summaries.sql b/warehouse/models/intermediate/gtfs/int_gtfs_rt__service_alerts_trip_summaries.sql
@@ -0,0 +1,53 @@
+{{
+    config(
+        materialized='incremental',
+        incremental_strategy='insert_overwrite',
+        partition_by={
+            'field': 'dt',
+            'data_type': 'date',
+            'granularity': 'day',
+        },
+        cluster_by='base64_url',
+    )
+}}
+
+{% if is_incremental() %}
+    {% set timestamps = dbt_utils.get_column_values(table=this, column='dt', order_by = 'dt DESC', max_records = 1) %}
+    {% set max_ts = timestamps[0] %}
+{% endif %}
+
+WITH service_alerts AS (
+    SELECT * FROM {{ ref('fct_service_alert_informed_entities') }}
+    {% if is_incremental() %}
+    WHERE dt >= EXTRACT(DATE FROM TIMESTAMP('{{ max_ts }}'))
+    {% else %}
+    WHERE dt >= DATE_SUB(CURRENT_DATE(), INTERVAL {{ var('TRIP_UPDATES_LOOKBACK_DAYS') }} DAY)
+    {% endif %}
+),
+
+int_gtfs_rt__service_alerts_trip_summaries AS (
+    SELECT
+        -- https://gtfs.org/realtime/reference/#message-tripdescriptor
+        {{ dbt_utils.surrogate_key([
+            'dt',
+            'base64_url',
+            'trip_id',
+            'trip_route_id',
+            'trip_direction_id',
+            'trip_start_time',
+            'trip_start_date',
+        ]) }} as key,
+        dt,
+        base64_url,
+        trip_id,
+        trip_route_id,
+        trip_direction_id,
+        trip_start_time,
+        trip_start_date,
+        COUNT(DISTINCT id) AS num_distinct_message_ids,
+        ARRAY_AGG(DISTINCT service_alert_message_key) AS service_alert_message_keys,
+    FROM service_alerts
+    GROUP BY 1, 2, 3, 4, 5, 6, 7, 8
+)
+
+SELECT * FROM int_gtfs_rt__service_alerts_trip_summaries
diff --git a/warehouse/models/intermediate/gtfs/int_gtfs_rt__trip_updates_summaries.sql b/warehouse/models/intermediate/gtfs/int_gtfs_rt__trip_updates_summaries.sql
@@ -12,7 +12,7 @@
 }}
 
 {% if is_incremental() %}
-    {% set timestamps = dbt_utils.get_column_values(table=this, column='dt', order_by = 'dt, DESC', max_records = 1) %}
+    {% set timestamps = dbt_utils.get_column_values(table=this, column='dt', order_by = 'dt DESC', max_records = 1) %}
     {% set max_ts = timestamps[0] %}
 {% endif %}
 

diff --git a/warehouse/models/intermediate/gtfs/int_gtfs_rt__vehicle_positions_trip_summaries.sql b/warehouse/models/intermediate/gtfs/int_gtfs_rt__vehicle_positions_trip_summaries.sql
@@ -12,7 +12,7 @@
 }}
 
 {% if is_incremental() %}
-    {% set timestamps = dbt_utils.get_column_values(table=this, column='dt', order_by = 'dt, DESC', max_records = 1) %}
+    {% set timestamps = dbt_utils.get_column_values(table=this, column='dt', order_by = 'dt DESC', max_records = 1) %}
     {% set max_ts = timestamps[0] %}
 {% endif %}
 

diff --git a/warehouse/models/mart/gtfs/fct_observed_trips.sql b/warehouse/models/mart/gtfs/fct_observed_trips.sql
@@ -19,6 +19,10 @@ vehicle_positions AS (
     SELECT * FROM {{ ref('int_gtfs_rt__vehicle_positions_trip_summaries') }}
 ),
 
+service_alerts AS (
+    SELECT * FROM {{ ref('int_gtfs_rt__service_alerts_trip_summaries') }}
+),
+
 fct_observed_trips AS (
     SELECT
         {{ dbt_utils.surrogate_key([
@@ -40,9 +44,13 @@ fct_observed_trips AS (
         vp.num_distinct_message_ids AS vp_num_distinct_message_ids,
         vp.min_trip_update_timestamp AS vp_min_trip_update_timestamp,
         vp.max_trip_update_timestamp AS vp_max_trip_update_timestamp,
+        sa.num_distinct_message_ids AS sa_num_distinct_message_ids,
+        sa.service_alert_message_keys AS sa_service_alert_message_key,
     FROM trip_updates AS tu
     FULL OUTER JOIN vehicle_positions AS vp
         USING (dt, base64_url, trip_id, trip_route_id, trip_direction_id, trip_start_time, trip_start_date)
+    FULL OUTER JOIN service_alerts AS sa
+        USING (dt, base64_url, trip_id, trip_route_id, trip_direction_id, trip_start_time, trip_start_date)
 )
 
 SELECT * FROM fct_observed_trips