Merge pull request #1924 from cal-itp/jan-open-data

Jan open data, partial progress

Author: edasmalchi

_shared_utils/shared_utils/rt_dates.py

@@ -89,6 +89,8 @@
     "oct2025": "2025-10-15",
     "oct2025b": "2025-10-16",
     "nov2025": "2025-11-05",
+    "dec2025": "2025-12-17",
+    "jan2026": "2026-01-14",
 }
 
 years_available = list(range(2023, datetime.datetime.now().year + 1))
@@ -100,6 +102,8 @@
 
 y2024_dates = [v for k, v in DATES.items() if k.endswith("2024") and k not in ["oct2024g"]]
 y2025_dates = [v for k, v in DATES.items() if k.endswith("2025")]
+y2026_dates = [v for k, v in DATES.items() if k.endswith("2026")]
+
 
 DATES_BY_YEAR_DICT = {
     2023: y2023_dates,

gtfs_funnel/Makefile

@@ -20,7 +20,7 @@ preprocess_vp:
 	python vp_condenser.py
 
 funnel_gtfs_single_day:
-	make download_gtfs_data
+#	make download_gtfs_data
 	make preprocess_schedule_vp_dependency
 	make preprocess_vp
 

gtfs_funnel/concatenate_vehicle_positions.py

@@ -1,24 +1,38 @@
 """
-Concatenate batched vehicle positions. 
+Concatenate batched vehicle positions.
 """
-import dask.dataframe as dd
-import dask_geopandas as dg
+
 import datetime
+import sys
+from functools import cache
+
+import dask.dataframe as dd
 import gcsfs
 import geopandas as gpd
 import pandas as pd
 import shapely
-import sys
-
-from dask import delayed, compute
+from calitp_data_analysis import utils
+from calitp_data_analysis.gcs_geopandas import GCSGeoPandas
+from calitp_data_analysis.gcs_pandas import GCSPandas
+from dask import compute, delayed
 from loguru import logger
-
 from shared_utils import schedule_rt_utils
-from calitp_data_analysis import utils
 from update_vars import GTFS_DATA_DICT, SEGMENT_GCS
 
+
+@cache
+def gcs_pandas():
+    return GCSPandas()
+
+
+@cache
+def gcs_geopandas():
+    return GCSGeoPandas()
+
+
 fs = gcsfs.GCSFileSystem()
 
+
 def concat_batches(analysis_date: str) -> dd.DataFrame:
     """
     Append individual operator vehicle position parquets together
@@ -27,19 +41,16 @@ def concat_batches(analysis_date: str) -> dd.DataFrame:
 
     fs_list = fs.ls(f"{SEGMENT_GCS}")
 
-    vp_files = [i for i in fs_list if "vp_raw" in i 
-                and f"{analysis_date}_batch" in i]
-    
-    delayed_dfs = [delayed(pd.read_parquet)(f"gs://{f}") 
-                   for f in vp_files]
-    
+    vp_files = [i for i in fs_list if "vp_raw" in i and f"{analysis_date}_batch" in i]
+
+    delayed_dfs = [delayed(gcs_pandas().read_parquet)(f"gs://{f}") for f in vp_files]
+
     ddf = dd.from_delayed(delayed_dfs)
-    
+
     ddf = schedule_rt_utils.localize_timestamp_col(
-        ddf, 
-        ["location_timestamp"] # add moving_timestamp with new mart table
+        ddf, ["location_timestamp"]  # add moving_timestamp with new mart table
     )
-    
+
     return ddf
 
 
@@ -49,53 +60,44 @@ def vp_into_gdf(df: pd.DataFrame) -> gpd.GeoDataFrame:
     """
     # Drop Nones or else shapely will error
     df2 = df[df.location.notna()].reset_index(drop=True)
-    
+
     geom = [shapely.wkt.loads(x) for x in df2.location]
 
-    gdf = gpd.GeoDataFrame(
-        df2, geometry=geom, 
-        crs="EPSG:4326").drop(columns="location")
-        
+    gdf = gpd.GeoDataFrame(df2, geometry=geom, crs="EPSG:4326").drop(columns="location")
+
     return gdf
 
 
 def remove_batched_parquets(analysis_date: str):
     """
-    Remove the batches of parquet downloads. 
+    Remove the batches of parquet downloads.
     These have file name pattern of *_batch*.
     """
     fs_list = fs.ls(f"{SEGMENT_GCS}")
-    
-    vp_files = [
-        i for i in fs_list if "vp_raw" in i 
-        and f"{analysis_date}_batch" in i
-    ]
-    
-    concat_file = [i for i in fs_list if 
-                   f"{analysis_date}_concat" in i 
-    ]
-    
+
+    vp_files = [i for i in fs_list if "vp_raw" in i and f"{analysis_date}_batch" in i]
+
+    concat_file = [i for i in fs_list if f"{analysis_date}_concat" in i]
+
     for f in vp_files:
         fs.rm(f)
-    
+
     for f in concat_file:
         fs.rm(f, recursive=True)
-       
-    
+
+
 if __name__ == "__main__":
-    
+
     from update_vars import analysis_date_list
 
     LOG_FILE = "./logs/download_vp_v2.log"
     logger.add(LOG_FILE, retention="3 months")
-    logger.add(sys.stderr, 
-               format="{time:YYYY-MM-DD at HH:mm:ss} | {level} | {message}", 
-               level="INFO")
-    
+    logger.add(sys.stderr, format="{time:YYYY-MM-DD at HH:mm:ss} | {level} | {message}", level="INFO")
+
     RAW_VP = GTFS_DATA_DICT.speeds_tables.raw_vp
-    
+
     for analysis_date in analysis_date_list:
-    
+
         logger.info(f"Analysis date: {analysis_date}")
 
         start = datetime.datetime.now()
@@ -106,36 +108,34 @@ def remove_batched_parquets(analysis_date: str):
         time1 = datetime.datetime.now()
         logger.info(f"concat and filter batched data: {time1 - start}")
 
-        concatenated_vp_df.to_parquet(
-            f"{SEGMENT_GCS}{RAW_VP}_{analysis_date}_concat", 
-            partition_on = "gtfs_dataset_key")
+        gcs_pandas().data_frame_to_parquet(
+            concatenated_vp_df, f"{SEGMENT_GCS}{RAW_VP}_{analysis_date}_concat", partition_on="gtfs_dataset_key"
+        )
 
         time2 = datetime.datetime.now()
         logger.info(f"export concatenated vp: {time2 - time1}")
 
         # Delete objects once it's saved out
         # Loop to save out multiple dates of vp may cause kernel to crash
         del concatenated_vp_df
-        
+
         # Import concatenated tabular vp and make it a gdf
-        vp = delayed(pd.read_parquet)(
-            f"{SEGMENT_GCS}{RAW_VP}_{analysis_date}_concat/"
-        ).reset_index(drop=True)
+        # Stripping gs:// from the supplied path avoids a pyarrow error?
+        logger.info(f"path: {SEGMENT_GCS[5:]}{RAW_VP}_{analysis_date}_concat/")
+        vp = delayed(gcs_pandas().read_parquet)(f"{SEGMENT_GCS[5:]}{RAW_VP}_{analysis_date}_concat/").reset_index(
+            drop=True
+        )
 
         vp_gdf = delayed(vp_into_gdf)(vp)
-        
+
         vp_gdf = compute(vp_gdf)[0]
 
-        utils.geoparquet_gcs_export(
-            vp_gdf,
-            SEGMENT_GCS,
-            f"{RAW_VP}_{analysis_date}"
-        )
+        utils.geoparquet_gcs_export(vp_gdf, SEGMENT_GCS, f"{RAW_VP}_{analysis_date}")
 
         remove_batched_parquets(analysis_date)
-        logger.info(f"remove batched parquets")
+        logger.info("remove batched parquets")
 
         end = datetime.datetime.now()
         logger.info(f"execution time: {end - start}")
-        
+
         del vp_gdf