fix(bus): regeneration of bus datasets hangs on old days (#667)

* temporary change to limit metrics regeneration to 2025 - OOM errors processing 2023 data.

* add fall2025 bus dataset to tableau job lists

* need partition template otherwise it loops through and processes the whole directory...many times over.

* need to do it for the original bus_mapping files as well

* set bus n days to include fall rating that started aug 24

* setting a date in write_bus_metrics causes it to always reprocess all of the files from that date forward. this is causing us to process all of 2025 constantly.

* remove import unused

Author: huangh

src/lamp_py/tableau/jobs/lamp_jobs.py

@@ -1,3 +1,4 @@
+from datetime import date
 import os
 from lamp_py.bus_performance_manager.events_joined import TMDailyWorkPiece
 from lamp_py.common.gtfs_types import RouteType
@@ -7,6 +8,9 @@
 )
 
 from lamp_py.runtime_utils.remote_files import (
+    LAMP,
+    S3_ARCHIVE,
+    S3Location,
     springboard_rt_vehicle_positions,
     springboard_devgreen_rt_vehicle_positions,
     springboard_rt_trip_updates,  # main feed, all lines, unique records
@@ -26,6 +30,13 @@
     tableau_rail_subway,
 )
 
+
+from lamp_py.runtime_utils.remote_files import (
+    bus_events,
+)
+from lamp_py.tableau.conversions.convert_bus_performance_data import apply_bus_analysis_conversions
+from lamp_py.tableau.jobs.bus_performance import bus_schema
+
 from lamp_py.tableau.jobs.filtered_hyper import FilteredHyperJob, days_ago
 from lamp_py.tableau.jobs.rt_rail import HyperRtRail
 from lamp_py.utils.filter_bank import FilterBankRtTripUpdates, FilterBankRtVehiclePositions
@@ -111,7 +122,7 @@
     dataframe_filter=None,
     parquet_filter=None,
     tableau_project_name=LAMP_API_PROJECT,
-    partition_template="",
+    partition_template="operator_map_pii_{yy}{mm:02d}{dd:02d}.parquet",
 )
 
 HyperBusOperatorMappingAll = FilteredHyperJob(
@@ -122,7 +133,39 @@
     dataframe_filter=None,
     parquet_filter=None,
     tableau_project_name=LAMP_API_PROJECT,
-    partition_template="",
+    partition_template="operator_map_pii_{yy}{mm:02d}{dd:02d}.parquet",
+)
+
+HyperBusFall2025 = FilteredHyperJob(
+    remote_input_location=bus_events,
+    remote_output_location=S3Location(
+        bucket=S3_ARCHIVE,
+        prefix=os.path.join(LAMP, "bus_rating_datasets", "year=2025", "Fall2025_BusMetrics.parquet"),
+        version="1.0",
+    ),
+    start_date=date(2025, 8, 24),
+    end_date=date(2025, 12, 13),
+    processed_schema=bus_schema,
+    dataframe_filter=apply_bus_analysis_conversions,
+    parquet_filter=None,
+    tableau_project_name=LAMP_API_PROJECT,
+    partition_template="{yy}{mm:02d}{dd:02d}.parquet",
+)
+
+HyperBusOperatorFall2025 = FilteredHyperJob(
+    remote_input_location=bus_operator_mapping,
+    remote_output_location=S3Location(
+        bucket=S3_ARCHIVE,
+        prefix=os.path.join(LAMP, "bus_rating_datasets", "year=2025", "Fall2025_Operator.parquet"),
+        version="1.0",
+    ),
+    start_date=date(2025, 8, 24),
+    end_date=date(2025, 12, 13),
+    processed_schema=TMDailyWorkPiece.to_pyarrow_schema(),
+    dataframe_filter=None,
+    parquet_filter=None,
+    tableau_project_name=LAMP_API_PROJECT,
+    partition_template="operator_map_pii_{yy}{mm:02d}{dd:02d}.parquet",
 )
 
 # light rail and heavy rail - Enum Types < 2 == 0, 1

src/lamp_py/tableau/pipeline.py

@@ -17,6 +17,8 @@
     HyperGtfsRtVehiclePositionsHeavyRail,
     HyperRtRailSubway,
     HyperRtRailCommuter,
+    HyperBusFall2025,
+    HyperBusOperatorFall2025,
 )
 from lamp_py.tableau.jobs.rt_alerts import HyperRtAlerts
 from lamp_py.tableau.jobs.gtfs_rail import (
@@ -91,6 +93,8 @@ def start_hyper_updates() -> None:
         HyperGtfsRtVehiclePositionsHeavyRail,
         HyperGtfsRtTripUpdatesHeavyRail,
         HyperGtfsRtVehiclePositionsAllLightRail,
+        HyperBusFall2025,
+        HyperBusOperatorFall2025,
     ]
 
     for job in hyper_jobs:
@@ -138,6 +142,8 @@ def start_bus_parquet_updates() -> None:
         HyperBusOperatorMappingRecent,
         HyperBusPerformanceAll(),
         HyperBusOperatorMappingAll,
+        HyperBusFall2025,
+        HyperBusOperatorFall2025,
     ]
 
     for job in parquet_update_jobs: