work new portfolio display names into merge_* scripts

Author: tiffanychu90

gtfs_digest/merge_data.py

@@ -306,7 +306,7 @@ def merge_data_sources_by_route_direction(
     df_rt_sched: pd.DataFrame,
     df_avg_speeds: pd.DataFrame,
     df_crosswalk: pd.DataFrame
-):
+) -> pd.DataFrame:
     """
     Merge schedule, rt_vs_schedule, and speeds data, 
     which are all at route-direction-time_period-date grain.

gtfs_digest/merge_operator_data.py

@@ -8,20 +8,23 @@
 
 from calitp_data_analysis import utils
 from segment_speed_utils import time_series_utils
-from shared_utils import gtfs_utils_v2, publish_utils
-from merge_data import merge_in_standardized_route_names
+from shared_utils import gtfs_utils_v2, portfolio_utils, publish_utils
+from merge_data import merge_in_standardized_route_names, PORTFOLIO_ORGANIZATIONS_DICT
 from update_vars import GTFS_DATA_DICT, SCHED_GCS, RT_SCHED_GCS
 
 sort_cols = ["schedule_gtfs_dataset_key", "service_date"]
 
 """
 Concatenating Functions 
 """
-def concatenate_rt_vs_schedule_operator_metrics(
+def concatenate_schedule_operator_metrics(
     date_list: list
 ) -> pd.DataFrame:
-    
-    FILE = f"{GTFS_DATA_DICT.rt_vs_schedule_tables.vp_operator_metrics}"
+    """
+    Get spatial accuracy and vehicle positions per minute metrics on the
+    operator-service_date grain for certain dates.
+    """
+    FILE = GTFS_DATA_DICT.schedule_tables.operator_scheduled_stats
 
     df = time_series_utils.concatenate_datasets_across_dates(
         SCHED_GCS,
@@ -32,9 +35,31 @@ def concatenate_rt_vs_schedule_operator_metrics(
 
     return df
 
+def concatenate_rt_vs_schedule_operator_metrics(
+    date_list: list
+) -> pd.DataFrame:
+    """
+    Concatenate operator grain RT vs schedule metrics
+    across all dates we have.
+    """
+    FILE = GTFS_DATA_DICT.rt_vs_schedule_tables.vp_operator_metrics
+    
+    df = time_series_utils.concatenate_datasets_across_dates(
+        RT_SCHED_GCS,
+        FILE,
+        date_list,
+        data_type = "df",
+    ).sort_values(sort_cols).reset_index(drop=True)
+    
+    return df
+
 def concatenate_operator_routes( 
     date_list: list
 ) -> gpd.GeoDataFrame:
+    """
+    Concatenate operator route gdf (1 representative shape chosen)
+    across all dates we have.
+    """
     FILE = GTFS_DATA_DICT.schedule_tables.operator_routes
 
     df = time_series_utils.concatenate_datasets_across_dates(
@@ -44,20 +69,24 @@ def concatenate_operator_routes(
         data_type = "gdf",
     ).sort_values(sort_cols).reset_index(drop=True)   
 
+    # TODO is there a short/long route, can it be flagged per date as a new column here?
+    
     return df
 
+
 def concatenate_crosswalks(
     date_list: list
 ) -> pd.DataFrame:
     """
     Get crosswalk and selected NTD columns for certain dates.
     """
-    FILE = f"{GTFS_DATA_DICT.schedule_tables.gtfs_key_crosswalk}"
+    FILE = GTFS_DATA_DICT.schedule_tables.gtfs_key_crosswalk
 
     ntd_cols = [
         "schedule_gtfs_dataset_key",
+        "name",
         "caltrans_district",
-        "counties_served",
+        #"counties_served", # remove this and create our own column
         "service_area_sq_miles",
         "hq_city",
         "service_area_pop",
@@ -74,91 +103,96 @@ def concatenate_crosswalks(
             data_type="df",
             columns=ntd_cols
         )
-        .sort_values(["service_date"])
+        .sort_values(sort_cols)
         .reset_index(drop=True)
+    ) 
+    
+    df = df.assign(
+        caltrans_district = df.caltrans_district.map(
+            portfolio_utils.CALTRANS_DISTRICT_DICT
+        )
+    ).pipe(
+        portfolio_utils.standardize_portfolio_organization_names, 
+        PORTFOLIO_ORGANIZATIONS_DICT
     )
 
+    
+    # to aggregate up to organization, 
+    # group by name-service_date-portfolio_organization_name
+    # because name indicates different feeds, so we want to sum those.
+    
+    
     return df
 
-def concatenate_schedule_operator_metrics(
-    date_list: list
+def merge_data_sources_by_operator(
+    df_schedule: pd.DataFrame,
+    df_rt_sched: pd.DataFrame,
+    df_crosswalk: pd.DataFrame
 ) -> pd.DataFrame:
     """
-    Get spatial accuracy and vehicle positions per minute metrics on the
-    operator-service_date grain for certain dates.
-    """
-    FILE = GTFS_DATA_DICT.schedule_tables.operator_scheduled_stats
-    
-    df = time_series_utils.concatenate_datasets_across_dates(
-        RT_SCHED_GCS,
-        FILE,
-        date_list,
-        data_type = "df",
-    ).sort_values(sort_cols).reset_index(drop=True)
-    
+    Merge schedule and rt_vs_schedule data, 
+    which are all at operator-date grain.
+    This merged dataset will be used in GTFS digest visualizations.
+    """    
+    df = pd.merge(
+        df_schedule,
+        df_rt_sched,
+        on = sort_cols,
+        how = "left",
+    ).merge(
+        df_crosswalk,
+        on = sort_cols + ["name"],
+        how = "inner"
+    )
+        
     return df
 
+## TODO: move counties stuff here
+# swap order at the bottom since this needs to be created first
+def counties_served_by_operator(route_gdf_by_operator):
+    """
+    take input produced in concatenate_operator_routes
+    get counties for operator-date
+    df should only be operator-date-counties_served
+    use this to merge into crosswalk and replace NTD column
+    """
+    
+    return
 
 if __name__ == "__main__":
 
     from shared_utils import rt_dates
 
     analysis_date_list = (
-        rt_dates.y2024_dates + rt_dates.y2023_dates +
-        rt_dates.y2025_dates
+        rt_dates.y2025_dates + rt_dates.y2024_dates + rt_dates.y2023_dates 
     )
 
     OPERATOR_PROFILE = GTFS_DATA_DICT.digest_tables.operator_profiles
     OPERATOR_ROUTE = GTFS_DATA_DICT.digest_tables.operator_routes_map
 
     public_feeds = gtfs_utils_v2.filter_to_public_schedule_gtfs_dataset_keys()
 
-    # Concat operator metrics.
-    op_sched_metrics = concatenate_schedule_operator_metrics(analysis_date_list)
-    
-    # Concat operator profiles
-    op_rt_sched_metrics = concatenate_rt_vs_schedule_operator_metrics(analysis_date_list)
-    
-    merge_cols = ["schedule_gtfs_dataset_key",
-             "service_date"]
-    
-    # Merge the two together
-    operator_profiles_df1 = pd.merge(op_sched_metrics, 
-                                  op_rt_sched_metrics,
-                                  on = merge_cols, 
-                                  how = "outer")
- 
+    # Concat operator grain for schedule metrics.
+    schedule_df = concatenate_schedule_operator_metrics(analysis_date_list)
 
+    # Concat operator grain for rt vs schedule metrics
+    rt_schedule_df = concatenate_rt_vs_schedule_operator_metrics(
+        analysis_date_list)
+
     # Concat NTD/crosswalk
     crosswalk_df = concatenate_crosswalks(analysis_date_list)
 
-    # Merge in NTD data. 
-    op_profiles_df2 = pd.merge(
-        operator_profiles_df1, 
-        crosswalk_df, 
-        on = merge_cols, 
-        how = "left"
+    operator_df = merge_data_sources_by_operator(
+        schedule_df,
+        rt_schedule_df,
+        crosswalk_df
+    ).pipe(
+        publish_utils.exclude_private_datasets, 
+        col = "schedule_gtfs_dataset_key", 
+        public_gtfs_dataset_keys = public_feeds
     )
 
-    # Drop duplicates created after merging
-    # Add more strigent drop duplicate criteria
-    duplicate_cols = ["schedule_gtfs_dataset_key",
-                     "vp_per_min_agency",
-                     "spatial_accuracy_agency",
-                     "service_date",
-                     "organization_name",
-                     "caltrans_district"]
-
-    op_profiles_df3 = (
-        op_profiles_df2
-        .pipe(
-            publish_utils.exclude_private_datasets, 
-            col = "schedule_gtfs_dataset_key", 
-            public_gtfs_dataset_keys = public_feeds
-        ).drop_duplicates(subset = duplicate_cols)
-    .reset_index(drop = True))
-
-    op_profiles_df3.to_parquet(
+    operator_df.to_parquet(
         f"{RT_SCHED_GCS}{OPERATOR_PROFILE}.parquet"
     )