Merge pull request #1420 from cal-itp/more-averaging

weekday-time_of_day-year segment speeds aggregation

Author: tiffanychu90

_shared_utils/shared_utils/gtfs_analytics_data.yml

@@ -62,8 +62,6 @@ rt_vs_schedule_tables:
 digest_tables:
   dir: ${gcs_paths.RT_SCHED_GCS}
   route_schedule_vp: "digest/schedule_vp_metrics"
-  route_segment_speeds: "digest/segment_speeds"
-  route_segment_geometry: "digest/segment_speeds_geom"
   operator_profiles: "digest/operator_profiles"
   operator_routes_map: "digest/operator_routes"
   operator_sched_rt: "digest/operator_schedule_rt_category"
@@ -85,7 +83,8 @@ rt_stop_times:
   trip_speeds_single_summary: "rollup_singleday/speeds_trip"
   route_dir_timeofday: "rollup_singleday/speeds_route_dir"
   segment_peakoffpeak_weekday_month: "rollup_singleday/month_speeds_route_dir_peak_segments" # NEW? what to name
-  segment_timeofday_weekday_year: "rollup_multiday/year_weekday_speeds_route_dir_segments"
+  segment_timeofday_weekday_year: "rollup_multiday/weekday_speeds_route_dir_segments"
+  segments_year_file: "rollup_multiday/stop_segments"
   min_trip_minutes: ${speed_vars.time_min_cutoff}
   max_trip_minutes: 180
   max_speed: ${speed_vars.max_speed}

_shared_utils/shared_utils/rt_dates.py

@@ -87,7 +87,7 @@
 valid_weeks = ["apr2023", "oct2023", "apr2024", "oct2024"]
 
 # Remove all the one-offs
-one_off_dates = ["jan2023", "feb2023" "aug2023a", "oct2024g"]
+one_off_dates = ["jan2023", "feb2023", "aug2023a", "oct2024g"]
 all_dates = [v for k, v in DATES.items() if k not in one_off_dates and "2022" not in k]
 
 

rt_segment_speeds/logs/avg_speeds.log

@@ -18,3 +18,6 @@
 2025-03-03 13:43:54.606 | INFO     | __main__:summary_speeds_by_peak_offpeak:158 - rt_stop_times summary speed averaging by peak/offpeak for 2025-02-12 execution time: 0:00:13.043791
 2025-03-03 13:44:02.598 | INFO     | __main__:trip_summary_speeds_by_time_of_day:92 - rt_stop_times summary speed averaging by time-of-day 2025-01-15 execution time: 0:00:07.986208
 2025-03-03 13:44:15.279 | INFO     | __main__:summary_speeds_by_peak_offpeak:158 - rt_stop_times summary speed averaging by peak/offpeak for 2025-01-15 execution time: 0:00:12.679323
+2025-03-14 13:34:30.155 | INFO     | __main__:annual_time_of_day_averages:171 - rt_stop_times: weekday/time-of-day averages for 2023 execution time: 0:02:23.782640
+2025-03-14 13:36:37.069 | INFO     | __main__:annual_time_of_day_averages:171 - rt_stop_times: weekday/time-of-day averages for 2024 execution time: 0:02:06.806331
+2025-03-14 13:36:50.516 | INFO     | __main__:annual_time_of_day_averages:171 - rt_stop_times: weekday/time-of-day averages for 2025 execution time: 0:00:13.404898

rt_segment_speeds/scripts/average_segment_speeds.py

@@ -14,6 +14,7 @@
 from typing import Literal, Optional
 
 from calitp_data_analysis import utils
+from calitp_data_analysis.geography_utils import WGS84
 
 from segment_speed_utils import gtfs_schedule_wrangling, segment_calcs, time_series_utils
 from shared_utils import publish_utils, time_helpers
@@ -43,8 +44,60 @@ def import_singleday_segment_speeds(
     return df
 
 
+def export_segment_geometry(
+    year: str,
+):
+    """
+    Dedupe segment geometries using columns, 
+    since geometries may slightly differ.
+    Visual inspection shows start and endpoints might be
+    slightly different but still capture the same corridor.
+    
+    Big Blue Bus: stop_pair = "1115__187"
+    In 2024, there are 4 rows, but the 4 rows are basically the same,
+    so let's keep the most recent row.
+    """
+    SEGMENTS_FILE = GTFS_DATA_DICT.rt_stop_times.segments_file
+    EXPORT_FILE = GTFS_DATA_DICT.rt_stop_times.segments_year_file
+    
+    keep_cols = [
+        "schedule_gtfs_dataset_key", 
+        "route_id", "direction_id", 
+        "stop_pair", 
+    ]
+    
+    dates_in_year = [
+        date for date in rt_dates.all_dates if year in date
+    ]
+    
+    df = time_series_utils.concatenate_datasets_across_dates(
+        SEGMENT_GCS,
+        SEGMENTS_FILE,
+        dates_in_year,
+        columns = keep_cols + ["geometry"],
+        data_type = "gdf",
+        get_pandas= False,        
+    ).sort_values(
+        "service_date", ascending=False
+    ).drop(
+        columns = "service_date"
+    ).drop_duplicates(
+        subset = keep_cols
+    ).reset_index(drop=True).to_crs(WGS84)
+
+    df = df.compute()
+
+    df.to_parquet(
+        f"{SEGMENT_GCS}{EXPORT_FILE}_{year}.parquet",
+    )
+    
+    print(f"exported stop segments for year {year}")
+        
+    return 
+    
+
 def annual_time_of_day_averages(
-    analysis_date_list: list,
+    year: str,
     segment_type: Literal[SEGMENT_TYPES],
     config_path: Optional = GTFS_DATA_DICT
 ):
@@ -63,6 +116,7 @@ def annual_time_of_day_averages(
     dict_inputs = config_path[segment_type]
 
     SPEED_FILE = dict_inputs["segment_timeofday"]
+    SEGMENTS_YEAR_FILE = dict_inputs["segments_year_file"]
     EXPORT_FILE = dict_inputs["segment_timeofday_weekday_year"]
 
     SEGMENT_COLS = [*dict_inputs["segment_cols"]]
@@ -71,6 +125,10 @@ def annual_time_of_day_averages(
     OPERATOR_COLS = ["schedule_gtfs_dataset_key"]
     CROSSWALK_COLS = [*dict_inputs.crosswalk_cols]
 
+    analysis_date_list = [
+        date for date in rt_dates.all_dates if year in date
+    ]
+    
     df = import_singleday_segment_speeds(
         SEGMENT_GCS, 
         SPEED_FILE, 
@@ -80,50 +138,44 @@ def annual_time_of_day_averages(
     ).pipe(
         time_helpers.add_quarter
     )
-       
-    avg_speeds = segment_calcs.calculate_weighted_averages(
-        df,
-        OPERATOR_COLS + SEGMENT_COLS_NO_GEOM + ["time_of_day", "weekday_weekend", "year"],
-        metric_cols = ["p20_mph", "p50_mph", "p80_mph"], 
-        weight_col = "n_trips"
-    ).persist()
 
-    publish_utils.if_exists_then_delete(
-        f"{SEGMENT_GCS}{EXPORT_FILE}"
-    )
+    group_cols = OPERATOR_COLS + SEGMENT_COLS_NO_GEOM + [
+        "time_of_day", "weekday_weekend", "year"]
 
-    avg_speeds.to_parquet(
-        f"{SEGMENT_GCS}{EXPORT_FILE}",
-        partition_on = "time_of_day"
-    )
-    '''
-    speeds_gdf = delayed(segment_calcs.merge_in_segment_geometry)(
-        avg_speeds,
-        analysis_date_list,
-        segment_type,
-        SEGMENT_COLS
-    ).pipe(
-        gtfs_schedule_wrangling.merge_operator_identifiers, 
+    speed_cols = ["p20_mph", "p50_mph", "p80_mph"]
+    weight_col = "n_trips"
+    
+    orig_dtypes = df[group_cols + speed_cols + [weight_col]].dtypes.to_dict()
+    
+    avg_speeds = df.map_partitions(
+        segment_calcs.calculate_weighted_averages,
+        OPERATOR_COLS + SEGMENT_COLS_NO_GEOM + ["time_of_day", "weekday_weekend", "year"],
+        metric_cols = speed_cols, 
+        weight_col = weight_col,
+        meta = {
+            **orig_dtypes,
+        },
+        align_dataframes = False
+    ).compute().pipe(
+        gtfs_schedule_wrangling.merge_operator_identifiers,
         analysis_date_list,
         columns = CROSSWALK_COLS
     )
 
-    utils.geoparquet_gcs_export(
-        speeds_gdf,
-        SEGMENT_GCS,
-        EXPORT_FILE
+    avg_speeds.to_parquet(
+        f"{SEGMENT_GCS}{EXPORT_FILE}_{year}.parquet"
     )
-    '''
 
     end = datetime.datetime.now()
 
     logger.info(
-        f"{segment_type}: weekday/time-of-day averages for {analysis_date_list} "
+        f"{segment_type}: weekday/time-of-day averages for {year} "
         f"execution time: {end - start}"
     )
 
     return 
 
+
 if __name__ == "__main__":
 
     from shared_utils import rt_dates 
@@ -135,9 +187,18 @@ def annual_time_of_day_averages(
                format="{time:YYYY-MM-DD at HH:mm:ss} | {level} | {message}", 
                level="INFO")
 
+    # isolate segments per year to allow for export
+    # rerun previous years when necessary
+    for year in ["2025"]:
+        
+        export_segment_geometry(year)
 
-    annual_time_of_day_averages(
-        rt_dates.all_dates,
-        segment_type = "rt_stop_times",
-    )
+        annual_time_of_day_averages(
+            year,
+            segment_type = "rt_stop_times",
+        )
+    
+
+        
+