Support LAMP based monthly csvs

mbta-performance/chalicelib/historic/backfill/main.py

@@ -1,4 +1,4 @@
-from ..constants import ARCGIS_IDS
+from ..constants import ARCGIS_IDS, HISTORIC_COLUMNS_LAMP, HISTORIC_COLUMNS_PRE_LAMP
 from ..download import download_historic_data, list_files_in_dir, prep_local_dir, unzip_historic_data
 from ..process import process_events
 
@@ -11,7 +11,11 @@ def backfill_single_year(year: str):
     input_dir = unzip_historic_data(zip_file, f"data/input/{year}")
     # process the data
     for file in list_files_in_dir(input_dir):
-        process_events(file, "data/output")
+        # in 2024 data moved to LAMP and the format changed
+        if int(year) >= 2024:
+            process_events(file, "data/output", columns=HISTORIC_COLUMNS_LAMP)
+        else:
+            process_events(file, "data/output", columns=HISTORIC_COLUMNS_PRE_LAMP)
     print(f"Finished backfilling year {year}")
 
 

mbta-performance/chalicelib/historic/constants.py

@@ -7,11 +7,11 @@
     "2021": "611b8c77f30245a0af0c62e2859e8b49",
     "2022": "99094a0c59e443cdbdaefa071c6df609",
     "2023": "9a7f5634db72459ab731b6a9b274a1d4",
-    "2024": "4adbec39db40498a8530496d8c63a924",
+    "2024": "0711756aa5e1400891e79b984a94b495",
 }
 
 
-HISTORIC_COLUMNS = [
+HISTORIC_COLUMNS_PRE_LAMP = [
     "service_date",
     "route_id",
     "trip_id",
@@ -23,3 +23,16 @@
     "event_type",
     "event_time_sec",
 ]
+
+HISTORIC_COLUMNS_LAMP = [
+    "service_date",
+    "route_id",
+    "trip_id",
+    "direction_id",
+    "stop_id",
+    "sync_stop_sequence",
+    "vehicle_id",
+    "vehicle_label",
+    "event_type",
+    "event_time_sec",
+]

mbta-performance/chalicelib/historic/process.py

@@ -1,13 +1,13 @@
 import pandas as pd
 import pathlib
-from .constants import HISTORIC_COLUMNS
+from .constants import HISTORIC_COLUMNS_PRE_LAMP as HISTORIC_COLUMNS
 from .gtfs_archive import add_gtfs_headways
 
 
-def process_events(input_csv: str, outdir: str, nozip: bool = False):
+def process_events(input_csv: str, outdir: str, nozip: bool = False, columns: list = HISTORIC_COLUMNS):
     df = pd.read_csv(
         input_csv,
-        usecols=HISTORIC_COLUMNS,
+        usecols=columns,
         parse_dates=["service_date"],
         dtype={
             "route_id": "str",
@@ -22,13 +22,24 @@ def process_events(input_csv: str, outdir: str, nozip: bool = False):
     df["event_time"] = df["service_date"] + pd.to_timedelta(df["event_time_sec"], unit="s")
     df.drop(columns=["event_time_sec"], inplace=True)
 
+    if "sync_stop_sequence" in df.columns:
+        df.rename(columns={"sync_stop_sequence": "stop_sequence"}, inplace=True)
+
     try:
         df = add_gtfs_headways(df)
     except IndexError:
         # failure to add gtfs benchmarks
         pass
 
-    service_date_month = pd.Grouper(key="service_date", freq="1M")
+    # Write to disk
+    to_disk(df, outdir, nozip)
+
+
+def to_disk(df: pd.DataFrame, outdir, nozip=False):
+    """
+    For each service_date/stop_id/direction/route group, we write the events to disk.
+    """
+    service_date_month = pd.Grouper(key="service_date", freq="1ME")
     grouped = df.groupby([service_date_month, "stop_id"])
 
     for name, events in grouped: