conveyal update gets old feeds if current feed only has future service defined

Author: mrtopsyt

conveyal_update/conveyal_vars.py

@@ -3,7 +3,9 @@
 
 GCS_PATH = 'gs://calitp-analytics-data/data-analyses/conveyal_update/'
 TARGET_DATE = rt_dates.DATES['mar2025']  
+LOOKBACK_TIME = dt.timedelta(days=60)
 OSM_FILE = 'us-west-latest.osm.pbf'
+PUBLISHED_FEEDS_YML_PATH = "../gtfs_funnel/published_operators.yml"
 #  http://download.geofabrik.de/north-america/us-west-latest.osm.pbf
 #  first download with wget...
 

conveyal_update/download_data.py

@@ -14,7 +14,6 @@
 regions = conveyal_vars.conveyal_regions
 TARGET_DATE = conveyal_vars.TARGET_DATE
 
-regions_and_feeds = pd.read_parquet(f'{conveyal_vars.GCS_PATH}regions_feeds_{TARGET_DATE}.parquet')
 
 def download_feed(row):
     # need wildcard for file too -- not all are gtfs.zip!
@@ -29,7 +28,8 @@ def download_region(feeds_df, region: str):
 
     assert region in regions.keys()
     path = f'./feeds_{feeds_df.date.iloc[0].strftime("%Y-%m-%d")}/{region}'
-    if not os.path.exists(path): os.makedirs(path)
+    if not os.path.exists(path): 
+        os.makedirs(path)
     region = (feeds_df >> filter(_.region == region)).copy()
     region['path'] = path
     region.progress_apply(download_feed, axis = 1)
@@ -46,6 +46,7 @@ def generate_script(regions):
         f.write('\n'.join(cmds))      
 
 if __name__ == '__main__':
+    regions_and_feeds = pd.read_parquet(f'{conveyal_vars.GCS_PATH}regions_feeds_{TARGET_DATE}.parquet')
 
     for region in tqdm(regions.keys()):
         download_region(regions_and_feeds, region)

conveyal_update/evaluate_feeds.py

@@ -3,12 +3,16 @@
 from shared_utils import gtfs_utils_v2
 
 from calitp_data_analysis.tables import tbls
+from calitp_data_analysis.sql import query_sql
 from siuba import *
 import pandas as pd
 import datetime as dt
 
 import conveyal_vars
 
+TARGET_DATE = conveyal_vars.TARGET_DATE
+REGIONAL_SUBFEED_NAME = "Regional Subfeed"
+
 def check_defined_elsewhere(row, df):
     '''
     for feeds without service defined, check if the same service is captured in another feed that does include service
@@ -17,11 +21,6 @@ def check_defined_elsewhere(row, df):
     row['service_any_feed'] = is_defined
     return row
 
-    
-
-TARGET_DATE = conveyal_vars.TARGET_DATE
-REGIONAL_SUBFEED_NAME = "Regional Subfeed"
-
 def get_feeds_check_service():
     feeds_on_target = gtfs_utils_v2.schedule_daily_feed_to_gtfs_dataset_name(selected_date=TARGET_DATE)
     feeds_on_target = feeds_on_target.rename(columns={'name':'gtfs_dataset_name'})
@@ -62,25 +61,115 @@ def attach_transit_services(feeds_on_target: pd.DataFrame):
     ].copy()
     return feeds_services_filtered
 
-def report_undefined(feeds_on_target: pd.DataFrame):
-    fname = 'no_apparent_service.csv'
+def get_undefined_feeds(feeds_on_target: pd.DataFrame) -> pd.DataFrame:
     undefined = feeds_on_target.apply(check_defined_elsewhere, axis=1, args=[feeds_on_target]) >> filter(-_.service_any_feed)
+    return undefined
+    
+INT_TO_GTFS_WEEKDAY = {
+    0: "monday",
+    1: "tuesday",
+    2: "wednesday",
+    3: "thursday",
+    4: "friday",
+    5: "saturday",
+    6: "sunday"
+}
+
+def report_unavailable_feeds(feeds, fname):
+    undefined = feeds.loc[
+        feeds["valid_date_other_than_service_date"] | ~feeds["usable_schedule_feed_exists"]
+    ].copy()
     if undefined.empty:
         print('no undefined service feeds')
     else:
-        print(undefined.columns)
         print('these feeds have no service defined on target date, nor are their services captured in other feeds:')
-        #  gtfs_dataset_name no longer present, this whole script should probably be updated/replaced
-        print(undefined >> select(_.gtfs_dataset_name, _.service_any_feed))
+        print(undefined.loc[~undefined["usable_schedule_feed_exists"], "gtfs_dataset_name"].drop_duplicates())
+        print('these feeds have defined service, but only in a feed defined on a prior day')
+        print(undefined.loc[undefined["valid_date_other_than_service_date"], "gtfs_dataset_name"].drop_duplicates())
         print(f'saving detailed csv to {fname}')
-        undefined.to_csv(fname)
-    return
+        undefined.to_csv(fname, index=False)
+
+ISO_DATE_ONLY_FORMAT = "%y-%m-%d"
+
+def get_old_feeds(undefined_feeds_base64_urls: pd.Series, target_date: dt.date | dt.datetime, max_lookback_timedelta: dt.timedelta) -> pd.Series:
+    
+    base_64_urls_str = "('" + "', '".join(undefined_feeds_base64_urls) + "')"
+    day_of_the_week = INT_TO_GTFS_WEEKDAY[target_date.weekday()]
+    max_lookback_date = target_date - max_lookback_timedelta
+    target_date_iso = target_date.strftime(ISO_DATE_ONLY_FORMAT)
+
+    query = f"""
+    SELECT 
+      `mart_gtfs.dim_schedule_feeds`.base64_url AS base64_url,
+      `mart_gtfs.dim_schedule_feeds`.key as feed_key,
+      `mart_gtfs.dim_calendar`.{day_of_the_week} AS target_day_of_the_week,
+      MAX(`mart_gtfs.dim_schedule_feeds`._valid_to) AS valid_feed_date,
+    from `mart_gtfs.dim_schedule_feeds`
+    LEFT JOIN `mart_gtfs.dim_calendar`
+    ON `mart_gtfs.dim_schedule_feeds`.key = `mart_gtfs.dim_calendar`.feed_key
+    WHERE `mart_gtfs.dim_schedule_feeds`.base64_url IN {base_64_urls_str}
+      AND `mart_gtfs.dim_schedule_feeds`._valid_to <= '{target_date}'
+      AND `mart_gtfs.dim_schedule_feeds`._valid_to >= '{max_lookback_date}'
+       AND `mart_gtfs.dim_calendar`.start_date <= '{target_date}'
+      AND `mart_gtfs.dim_calendar`.end_date >= '{target_date}'
+    GROUP BY 
+        `mart_gtfs.dim_schedule_feeds`.base64_url, 
+        `mart_gtfs.dim_schedule_feeds`.key,
+        `mart_gtfs.dim_calendar`.{day_of_the_week}
+    ORDER BY target_day_of_the_week DESC
+    LIMIT 1000
+    """
+    response = query_sql(
+        query
+    )
+    response_grouped = response.groupby("base64_url")
+    feed_info_by_url = response_grouped[["valid_feed_date", "feed_key"]].first()
+    print(feed_info_by_url)
+    feed_info_by_url["valid_feed_date"] = feed_info_by_url["valid_feed_date"].dt.date - dt.timedelta(days=1) 
+    # we have the day the feed becomes invalid, so the day we are interested in where the feed *is* valid is the day after
+    feed_info_by_url["no_operations_on_target_day_of_the_week"] = ~(response_grouped["target_day_of_the_week"].any())
+    return feed_info_by_url
+
+def merge_old_feeds(df_all_feeds: pd.DataFrame, df_undefined_feeds: pd.DataFrame, target_date: dt.date, max_lookback_timedelta: dt.timedelta) -> pd.DataFrame:
+    feed_search_result = get_old_feeds(
+        df_undefined_feeds["base64_url"], 
+        target_date,
+        max_lookback_timedelta
+    )
+    feeds_merged = df_all_feeds.merge(
+        feed_search_result,
+        how="left",
+        left_on="base64_url",
+        right_index=True,
+        validate="many_to_one"
+    ) 
+    feeds_merged["feed_key"] = feeds_merged["feed_key_y"].fillna(feeds_merged["feed_key_x"])
+    feeds_merged["no_schedule_feed_found"] = (
+        (feeds_merged["base64_url"].isin(df_undefined_feeds["base64_url"])) & (~feeds_merged["base64_url"].isin(feed_search_result.index))
+    )
+    feeds_merged["no_operations_on_target_date_but_valid_feed_exists"] = (feeds_merged["no_operations_on_target_day_of_the_week"].fillna(False))
+    feeds_merged["usable_schedule_feed_exists"] = (
+        ~(feeds_merged["no_schedule_feed_found"] | feeds_merged["no_operations_on_target_date_but_valid_feed_exists"])
+    )
+    feeds_merged["date"] = feeds_merged.loc[
+        ~feeds_merged["no_operations_on_target_date_but_valid_feed_exists"], "valid_feed_date"
+    ]
+    feeds_merged["date"] = feeds_merged["date"].fillna(target_date)
+    feeds_merged["valid_date_other_than_service_date"] = feeds_merged["date"] != target_date
+
+    return feeds_merged.drop(
+        ["valid_feed_date", "no_operations_on_target_day_of_the_week", "feed_key_x", "feed_key_y"], axis=1
+    )
 
 if __name__ == '__main__':
 
     feeds_on_target = get_feeds_check_service()
     feeds_on_target = attach_transit_services(feeds_on_target)
     print(f'feeds on target date shape: {feeds_on_target.shape}')
-    report_undefined(feeds_on_target)
-    feeds_on_target.to_parquet(f'{conveyal_vars.GCS_PATH}feeds_{TARGET_DATE}.parquet')
+    undefined_feeds = get_undefined_feeds(feeds_on_target)
+    feeds_merged = merge_old_feeds(
+        feeds_on_target, undefined_feeds, dt.date.fromisoformat(TARGET_DATE), conveyal_vars.LOOKBACK_TIME
+    )
+    report_unavailable_feeds(feeds_merged, 'no_apparent_service.csv')
+    feeds_merged.to_parquet(f'{conveyal_vars.GCS_PATH}feeds_{TARGET_DATE}.parquet')
 

conveyal_update/match_feeds_regions.py

@@ -13,7 +13,6 @@
 
 regions = conveyal_vars.conveyal_regions
 TARGET_DATE = conveyal_vars.TARGET_DATE
-feeds_on_target = pd.read_parquet(f'{conveyal_vars.GCS_PATH}feeds_{TARGET_DATE}.parquet')
 
 def create_region_gdf():
     # https://shapely.readthedocs.io/en/stable/reference/shapely.box.html#shapely.box
@@ -26,14 +25,24 @@ def create_region_gdf():
     region_gdf = gpd.GeoDataFrame(df, crs=geography_utils.WGS84).to_crs(geography_utils.CA_NAD83Albers_m)
     return region_gdf
 
+def get_stops_dates(feeds_on_target: pd.DataFrame, feed_key_column_name: str = "feed_key", date_column_name: str = "date"):
+    all_stops = feeds_on_target.groupby(date_column_name)[feed_key_column_name].apply(
+        lambda feed_key_column: gtfs_utils_v2.get_stops(
+            selected_date=feed_key_column.name,
+            operator_feeds=feed_key_column
+        )
+    )
+    return all_stops
+
 def join_stops_regions(region_gdf: gpd.GeoDataFrame, feeds_on_target: pd.DataFrame):
-    all_stops = gtfs_utils_v2.get_stops(selected_date=TARGET_DATE, operator_feeds=feeds_on_target.feed_key).to_crs(geography_utils.CA_NAD83Albers_m)
+    #all_stops = gtfs_utils_v2.get_stops(selected_date=TARGET_DATE, operator_feeds=feeds_on_target.feed_key)
+    all_stops = get_stops_dates(feeds_on_target).to_crs(geography_utils.CA_NAD83Albers_m)
     region_join = gpd.sjoin(region_gdf, all_stops)
     regions_and_feeds = region_join >> distinct(_.region, _.feed_key)
     return regions_and_feeds
 
 if __name__ == '__main__':
-    
+    feeds_on_target = pd.read_parquet(f'{conveyal_vars.GCS_PATH}feeds_{TARGET_DATE}.parquet')
     region_gdf = create_region_gdf()
     regions_and_feeds = join_stops_regions(region_gdf, feeds_on_target)
     regions_and_feeds = regions_and_feeds >> inner_join(_, feeds_on_target >> select(_.feed_key, _.gtfs_dataset_name, _.base64_url,