Test out LAMP for bus

mbta-performance/.chalice/config.json

@@ -25,7 +25,17 @@
         },
         "process_yesterday_lamp": {
           "iam_policy_file": "policy-lamp-ingest.json",
-          "lambda_timeout": 900,
+          "lambda_timeout": 600,
+          "lambda_memory_size": 4096
+        },
+        "process_daily_bus_lamp": {
+          "iam_policy_file": "policy-bus-lamp-ingest.json",
+          "lambda_timeout": 120,
+          "lambda_memory_size": 2048
+        },
+        "process_yesterday_bus_lamp": {
+          "iam_policy_file": "policy-bus-lamp-ingest.json",
+          "lambda_timeout": 600,
           "lambda_memory_size": 4096
         },
         "regenerate_tm_benchmarks": {
@@ -36,4 +46,4 @@
       }
     }
   }
-}
+}

mbta-performance/.chalice/policy-bus-lamp-ingest.json

@@ -0,0 +1,19 @@
+{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Action": [
+        "logs:CreateLogGroup",
+        "logs:CreateLogStream",
+        "logs:PutLogEvents"
+      ],
+      "Effect": "Allow",
+      "Resource": "arn:*:logs:*:*:*"
+    },
+    {
+      "Action": ["s3:GetObject", "s3:PutObject"],
+      "Effect": "Allow",
+      "Resource": ["arn:aws:s3:::tm-mbta-performance/Events-lamp/bus-daily-data/*"]
+    }
+  ]
+}

mbta-performance/app.py

@@ -46,3 +46,23 @@ def process_yesterday_lamp(event):
 def regenerate_tm_benchmarks(event):
     """Regenerate TransitMatters travel-time benchmarks for rapid transit."""
     benchmarks.generate_travel_time_benchmarks()
+
+
+# Bus LAMP data processing
+# Runs every 30 minutes from either 5 AM -> 2:30AM or 6 AM -> 3:30 AM depending on DST
+@app.schedule(Cron("*/30", "0-7,10-23", "*", "*", "?", "*"))
+def process_daily_bus_lamp(event):
+    """Ingest today's bus LAMP data."""
+    now_boston = datetime.now(ZoneInfo("US/Eastern"))
+
+    if now_boston.hour >= 3 and now_boston.hour < 6:
+        return
+
+    lamp.ingest_today_bus_data()
+
+
+# Runs once the next day at 11am or 12pm depending on DST
+@app.schedule(Cron("0", "15", "*", "*", "?", "*"))
+def process_yesterday_bus_lamp(event):
+    """Process yesterday's bus LAMP data, to ensure we have everything we need."""
+    lamp.ingest_yesterday_bus_data()

mbta-performance/chalicelib/lamp/__init__.py

@@ -1,3 +1,9 @@
-__all__ = ["ingest_today_lamp_data", "ingest_yesterday_lamp_data"]
+__all__ = [
+    "ingest_today_lamp_data",
+    "ingest_yesterday_lamp_data",
+    "ingest_today_bus_data",
+    "ingest_yesterday_bus_data",
+]
 
+from .bus_ingest import ingest_today_bus_data, ingest_yesterday_bus_data
 from .ingest import ingest_today_lamp_data, ingest_yesterday_lamp_data

mbta-performance/chalicelib/lamp/backfill/bus.py

@@ -0,0 +1,76 @@
+import logging
+import os
+from datetime import date, timedelta
+
+import pandas as pd
+
+from ... import parallel
+from ..bus_ingest import RTE_DIR_STOP, fetch_bus_pq_file_from_remote, ingest_bus_pq_file, upload_bus_to_s3
+
+logger = logging.getLogger(__name__)
+
+_parallel_upload = parallel.make_parallel(upload_bus_to_s3)
+
+EARLIEST_BUS_LAMP_DATA = date(2020, 1, 8)
+
+LOCAL_ARCHIVE_PATH = os.environ.get("LOCAL_ARCHIVE_PATH", "./feeds")
+
+
+def backfill_all_bus_dates(start_date: date = EARLIEST_BUS_LAMP_DATA):
+    """Backfill all dates with bus LAMP data, most recent to oldest."""
+
+    if start_date < EARLIEST_BUS_LAMP_DATA:
+        raise ValueError(f"start_date {start_date} is before earliest available data {EARLIEST_BUS_LAMP_DATA}")
+
+    end_date = date.today() - timedelta(days=1)
+    dates = pd.date_range(start_date, end_date, freq="d")
+    num_days = len(dates)
+    estimated_cost = num_days / 30
+
+    confirmation = input(
+        f"This will backfill {num_days} days of bus LAMP data ({start_date} to {end_date}). "
+        f"Estimated cost: ~${estimated_cost:.2f} (based on ~$1 per 30 days). "
+        "This will take hours. Are you sure you want to proceed? (yes/no): "
+    )
+    if confirmation.lower() not in ("yes", "y"):
+        print("You must enter 'yes' to proceed. Exiting.")
+        exit(1)
+
+    for backfill_timestamp in dates[::-1]:
+        date_to_backfill = backfill_timestamp.date()
+        try:
+            pq_df = fetch_bus_pq_file_from_remote(date_to_backfill)
+        except ValueError as e:
+            logger.warning(f"Failed to fetch {date_to_backfill}: {e}")
+            continue
+
+        logger.info(f"Processing {date_to_backfill}")
+        processed = ingest_bus_pq_file(pq_df, date_to_backfill, local_archive_path=LOCAL_ARCHIVE_PATH)
+
+        group_event_groups = processed.groupby(RTE_DIR_STOP)
+        logger.info(f"Uploading events for {len(group_event_groups)} route-direction-stop groups to S3")
+        _parallel_upload(group_event_groups, date_to_backfill)
+        logger.info(f"Finished {date_to_backfill}")
+
+
+if __name__ == "__main__":
+    import argparse
+    import os
+    from datetime import datetime
+
+    parser = argparse.ArgumentParser(description="Backfill bus LAMP data from a start date up to yesterday.")
+    parser.add_argument(
+        "--start-date",
+        type=lambda s: datetime.strptime(s, "%Y-%m-%d").date(),
+        default=EARLIEST_BUS_LAMP_DATA,
+        help=f"Start date in YYYY-MM-DD format (default: {EARLIEST_BUS_LAMP_DATA}).",
+    )
+    args = parser.parse_args()
+
+    log_level = os.environ.get("LOG_LEVEL", "INFO").upper()
+    logging.basicConfig(
+        level=getattr(logging, log_level, logging.INFO),
+        format="%(asctime)s - %(levelname)s: %(message)s",
+        datefmt="%H:%M:%S",
+    )
+    backfill_all_bus_dates(start_date=args.start_date)

mbta-performance/chalicelib/lamp/bus_constants.py

@@ -0,0 +1,56 @@
+# Remote URL for fetching bus LAMP data (daily parquet files)
+BUS_DAILY_URL_TEMPLATE = "https://performancedata.mbta.com/lamp/bus_vehicle_events/{YYYYMMDD}.parquet"
+
+# Columns to read from the source parquet files
+BUS_LAMP_COLUMNS = [
+    "service_date",
+    "route_id",
+    "trip_id",
+    "stop_id",
+    "direction_id",
+    "stop_sequence",
+    "vehicle_label",
+    "previous_stop_id",
+    # Actual timestamps (UTC-aware)
+    "stop_arrival_dt",
+    "stop_departure_dt",
+    # Actual metrics
+    "travel_time_seconds",
+    "stopped_duration_seconds",
+    "route_direction_headway_seconds",
+    # Scheduled metrics
+    "plan_travel_time_seconds",
+    "plan_route_direction_headway_seconds",
+]
+
+# Columns output to S3 events.csv
+BUS_S3_COLUMNS = [
+    "service_date",
+    "route_id",
+    "trip_id",
+    "direction_id",
+    "stop_id",
+    "stop_sequence",
+    "vehicle_label",
+    "event_type",
+    "event_time",
+    "travel_time_seconds",
+    "dwell_time_seconds",
+    "headway_seconds",
+    "scheduled_tt",
+    "scheduled_headway",
+]
+
+BUS_COLUMN_RENAME_MAP = {
+    "stopped_duration_seconds": "dwell_time_seconds",
+    "route_direction_headway_seconds": "headway_seconds",
+    "plan_travel_time_seconds": "scheduled_tt",
+    "plan_route_direction_headway_seconds": "scheduled_headway",
+}
+
+# Output S3 bucket and key template
+BUS_S3_BUCKET = "tm-mbta-performance"
+# month and day are not zero-padded
+BUS_S3_KEY_TEMPLATE = (
+    "Events-lamp/bus-daily-data/{route_id}-{direction_id}-{stop_id}/Year={YYYY}/Month={_M}/Day={_D}/events.csv"
+)