Merge pull request #1455 from cal-itp/vp-path-speeds

tiffanychu90 · web-flow · commit c01b1042e878 · 2025-04-18T09:08:39.000-07:00
vp modeled path to speeds
diff --git a/rt_segment_speeds/scripts/model_speeds_every100.py b/rt_segment_speeds/scripts/model_speeds_every100.py
@@ -0,0 +1,198 @@
+import datetime
+import numpy as np
+import pandas as pd
+import geopandas as gpd
+
+from segment_speed_utils import helpers, vp_transform
+from segment_speed_utils.project_vars import SEGMENT_GCS, GTFS_DATA_DICT, PROJECT_CRS
+
+from shared_utils import rt_dates
+from shared_utils.rt_utils import MPH_PER_MPS
+from project_condense_resample import project_point_onto_shape
+
+analysis_date = rt_dates.DATES["oct2024"]
+
+def grab_arrays_by_trip(df, meters_interval: int):
+    
+    intervaled_cutoffs = []
+    speed_series = []
+    
+    for row in df.itertuples():
+        
+        one_trip_distance_arr = getattr(row, "interpolated_distances")
+        one_trip_timestamp_arr = getattr(row, "resampled_timestamps")
+        
+        start_dist = int(np.floor(one_trip_distance_arr).min())
+        end_dist = int(np.ceil(one_trip_distance_arr).max())
+
+        intervaled_distance_cutoffs = np.array(range(start_dist, end_dist, meters_interval))
+        
+        speeds_for_trip = get_speeds_every_interval(
+            one_trip_distance_arr, 
+            one_trip_timestamp_arr,
+            intervaled_distance_cutoffs
+        )
+        
+        intervaled_cutoffs.append(intervaled_distance_cutoffs)
+        speed_series.append(speeds_for_trip)
+    
+    df2 = df.assign(
+        intervaled_meters = intervaled_cutoffs,
+        speeds = speed_series
+    )[["trip_instance_key", "intervaled_meters", "speeds"]]
+    
+    return df2
+    
+    
+def get_speeds_every_interval(
+    one_trip_distance_arr, 
+    one_trip_timestamp_arr,
+    intervaled_distance_cutoffs,
+):
+    
+    one_trip_speed_series = []
+
+    for i in range(0, len(intervaled_distance_cutoffs) - 1):
+        cut1 = intervaled_distance_cutoffs[i]
+        cut2 = intervaled_distance_cutoffs[i+1]
+        subset_indices = np.where((one_trip_distance_arr >= cut1) & (one_trip_distance_arr <= cut2))
+
+        subset_distances = one_trip_distance_arr[subset_indices]
+        subset_times = one_trip_timestamp_arr[subset_indices]
+
+        # should deltas be returned?
+        if len(subset_distances > 0):
+            one_speed = (
+                (subset_distances.max() - subset_distances.min()) / 
+                (subset_times.max() - subset_times.min()) 
+                * MPH_PER_MPS
+            )
+
+            one_trip_speed_series.append(one_speed)
+        else:
+            one_trip_speed_series.append(np.nan)
+    return one_trip_speed_series
+
+
+def grab_arrays_by_trip2(
+    df, 
+    distance_type = "",
+    intervaled_distance_column_or_meters = ""
+):
+    
+    intervaled_cutoffs = []
+    speed_series = []
+    
+    for row in df.itertuples():
+        
+        one_trip_distance_arr = getattr(row, "interpolated_distances")
+        one_trip_timestamp_arr = getattr(row, "resampled_timestamps")
+        should_calculate = np.array(getattr(row, "stop_meters_increasing"))
+
+        
+        start_dist = int(np.floor(one_trip_distance_arr).min())
+        end_dist = int(np.ceil(one_trip_distance_arr).max())        
+        
+        if distance_type == "equal_intervals":
+            intervaled_distance_cutoffs = np.array(
+                range(start_dist, end_dist, intervaled_distance_column_or_meters))
+
+        elif distance_type == "stop_to_stop":
+            intervaled_distance_cutoffs = getattr(row, intervaled_distance_column_or_meters)
+            #do_not_calculate_indices = np.where(should_calculate == False)[0]
+
+        speeds_for_trip = get_speeds_every_interval(
+            one_trip_distance_arr, 
+            one_trip_timestamp_arr,
+            intervaled_distance_cutoffs,
+        )
+        
+        #if len(do_not_calculate_indices) > 0:
+        #    speeds_for_trip[do_not_calculate_indices] = np.nan
+        
+        
+        if distance_type == "equal_intervals":
+            intervaled_cutoffs.append(intervaled_distance_cutoffs)
+            keep_cols = ["intervaled_meters", "speeds"]
+        elif distance_type == "stop_to_stop":
+            keep_cols = ["speeds", "stop_sequence"]
+        
+        speed_series.append(speeds_for_trip)
+
+    if distance_type == "equal_intervals":
+        df2 = df.assign(
+            intervaled_meters = intervaled_cutoffs,
+            speeds = speed_series
+        )
+        
+    elif distance_type == "stop_to_stop":
+        df2 = df.assign(
+            speeds = speed_series
+        )
+    
+    return df2[["trip_instance_key"] + keep_cols]
+
+    
+
+
+if __name__ == "__main__":
+    '''
+    for b in ["batch0", "batch1"]:
+        start = datetime.datetime.now()
+
+        meters_interval = 250
+        df = pd.read_parquet(
+            f"{SEGMENT_GCS}vp_condensed/vp_resampled_{b}_{analysis_date}.parquet",
+        )
+
+        results = grab_arrays_by_trip(df, meters_interval)
+        results.to_parquet(
+            f"{SEGMENT_GCS}rough_speeds_{meters_interval}m_{b}_{analysis_date}.parquet"
+        )
+
+        end = datetime.datetime.now()
+        print(f"{b} speeds every {meters_interval}m: {end - start}")
+    
+    
+    #batch0 speeds every 100m: 0:03:00.469936
+    #batch1 speeds every 100m: 0:02:50.197037
+    #batch0 speeds every 250m: 0:01:32.080767
+    #batch1 speeds every 250m: 0:01:38.365538
+    #batch0 speeds every stop: 0:01:05.459700    
+    #batch1 speeds every stop: 0:00:46.450538    
+    '''
+
+    for b in ["batch0", "batch1"]:
+        start = datetime.datetime.now()
+
+        df = pd.read_parquet(
+            f"{SEGMENT_GCS}vp_condensed/vp_resampled_{b}_{analysis_date}.parquet",
+        )
+        
+        subset_trips = df.trip_instance_key.unique().tolist()
+
+        stop_time_cutoffs = pd.read_parquet(
+            f"{SEGMENT_GCS}stop_times_projected_{analysis_date}.parquet",
+            filters = [[("trip_instance_key", "in", subset_trips)]],
+            columns = ["trip_instance_key", "stop_sequence", "stop_meters", "stop_meters_increasing"]
+        )
+        
+        gdf = pd.merge(
+            df,
+            stop_time_cutoffs,
+            on = "trip_instance_key",
+            how = "inner"
+        )
+
+        results = grab_arrays_by_trip2(
+            gdf,
+            distance_type = "stop_to_stop",
+            intervaled_distance_column_or_meters = "stop_meters",
+        )
+        
+        results.to_parquet(
+            f"{SEGMENT_GCS}rough_speeds_stop_to_stop_{b}_{analysis_date}.parquet"
+        )
+
+        end = datetime.datetime.now()
+        print(f"{b} speeds every stop: {end - start}")
diff --git a/rt_segment_speeds/scripts/stop_times_prep.py b/rt_segment_speeds/scripts/stop_times_prep.py
@@ -0,0 +1,110 @@
+import dask.dataframe as dd
+import dask_geopandas as dg
+import datetime
+import numpy as np
+import pandas as pd
+import geopandas as gpd
+
+from segment_speed_utils import helpers, vp_transform
+from segment_speed_utils.project_vars import SEGMENT_GCS, GTFS_DATA_DICT, PROJECT_CRS
+
+from shared_utils import rt_dates
+from resample import project_point_onto_shape
+
+analysis_date = rt_dates.DATES["oct2024"]
+
+
+def is_monotonically_increasing(my_array: list):
+    """
+    Somehow store whether projecting stop position onto vp path is increasing or not.
+    results that are True, we can calculate speeds for, otherwise, we shouldn't.
+    these results are better than stop_meters calculated off of shape.
+    """
+    my_array2 = np.array(my_array)
+    boolean_results = np.diff(my_array2) > 0
+    # add first observation, which is true because the first distance is compared to 0?
+    return np.array(boolean_results) #np.array([True] + boolean_results) 
+
+
+def merge_stop_times_and_vp(analysis_date: str):
+    stop_times = helpers.import_scheduled_stop_times(
+        analysis_date,
+        columns = ["trip_instance_key", "stop_sequence", "geometry"],
+        with_direction = True,
+        get_pandas = False,
+    )
+    
+    vp_path =gpd.read_parquet(
+        f"{SEGMENT_GCS}vp_condensed/vp_projected_{analysis_date}.parquet",
+        columns = ["trip_instance_key", "vp_geometry"]
+    ).drop_duplicates()
+    
+    stop_times_vp_geom = dd.merge(
+        stop_times,
+        vp_path,
+        on = "trip_instance_key",
+        how = "inner"
+    ).sort_values(["trip_instance_key", "stop_sequence"]).reset_index(drop=True)
+    
+    stop_times_vp_geom = stop_times_vp_geom.repartition(npartitions=20)
+    
+    return stop_times_vp_geom
+
+def project_stop_onto_vp_geom_and_condense(gddf):
+    
+    orig_dtypes = gddf.dtypes.to_dict()
+
+    gdf2 = gddf.map_partitions(
+        project_point_onto_shape,
+            line_geom = "vp_geometry", 
+            point_geom = "geometry",
+        meta = {
+            **orig_dtypes, 
+            "projected_meters": "float"
+        },
+        align_dataframes = True
+    )
+
+    gdf2 = gdf2.rename(
+        columns = {"projected_meters": "stop_meters"}
+    ).drop(
+        columns = ["vp_geometry", "geometry"]
+    ).persist()
+        
+    
+    return gdf2
+
+
+
+
+if __name__ == "__main__":
+    
+    
+    start = datetime.datetime.now()
+    
+    stop_times_vp_geom = merge_stop_times_and_vp(analysis_date)
+    
+    gdf = project_stop_onto_vp_geom_and_condense(stop_times_vp_geom).compute()
+    
+    results = (gdf
+        .groupby("trip_instance_key", group_keys=False)
+        .agg({
+            "stop_sequence": lambda x: list(x),
+            "stop_meters": lambda x: list(x)
+        })
+        .reset_index()
+       )
+
+    results = results.assign(
+        stop_meters_increasing = results.apply(
+            lambda x: is_monotonically_increasing(x.stop_meters), axis=1
+        )
+    )
+     
+    results.to_parquet(
+        f"{SEGMENT_GCS}stop_times_projected_{analysis_date}.parquet"
+    )
+
+    
+    end = datetime.datetime.now()
+    print(f"stop times prep: {end - start}")
