Compute days_ago at runtime (#685)

Author: runkelcorey

runners/run_gtfs_rt_parquet_converter.py

@@ -6,7 +6,7 @@
 from lamp_py.bus_performance_manager.events_joined import TMDailyWorkPiece
 from lamp_py.tableau.conversions import convert_gtfs_rt_trip_updates, convert_gtfs_rt_vehicle_position
 from lamp_py.tableau.hyper import HyperJob
-from lamp_py.tableau.jobs.filtered_hyper import FilteredHyperJob, days_ago
+from lamp_py.tableau.jobs.filtered_hyper import FilteredHyperJob
 from lamp_py.tableau.pipeline import (
     GTFS_RT_TABLEAU_PROJECT,
     HyperGtfsRtVehiclePositions,
@@ -87,7 +87,7 @@ def start_gtfs_rt_parquet_updates_local() -> None:
     HyperGtfsRtVehiclePositions = FilteredHyperJob(
         remote_input_location=springboard_rt_vehicle_positions,
         remote_output_location=tableau_rt_vehicle_positions_lightrail_60_day,
-        start_date=days_ago(60),
+        start_date=60,
         processed_schema=convert_gtfs_rt_vehicle_position.LightRailTerminalVehiclePositions.to_pyarrow_schema(),
         dataframe_filter=convert_gtfs_rt_vehicle_position.lrtp,
         parquet_filter=FilterBankRtVehiclePositions.ParquetFilter.light_rail,
@@ -97,7 +97,7 @@ def start_gtfs_rt_parquet_updates_local() -> None:
     HyperGtfsRtTripUpdates = FilteredHyperJob(
         remote_input_location=springboard_rt_trip_updates,
         remote_output_location=tableau_rt_trip_updates_lightrail_60_day,
-        start_date=days_ago(60),
+        start_date=60,
         processed_schema=convert_gtfs_rt_trip_updates.LightRailTerminalTripUpdates.to_pyarrow_schema(),
         dataframe_filter=convert_gtfs_rt_trip_updates.lrtp_prod,
         parquet_filter=FilterBankRtTripUpdates.ParquetFilter.light_rail,
@@ -107,7 +107,7 @@ def start_gtfs_rt_parquet_updates_local() -> None:
     HyperGtfsRtVehiclePositionsHeavyRail = FilteredHyperJob(
         remote_input_location=springboard_rt_vehicle_positions,
         remote_output_location=tableau_rt_vehicle_positions_heavyrail_30_day,
-        start_date=days_ago(60),
+        start_date=60,
         processed_schema=convert_gtfs_rt_vehicle_position.HeavyRailTerminalVehiclePositions.to_pyarrow_schema(),
         dataframe_filter=convert_gtfs_rt_vehicle_position.heavyrail,
         parquet_filter=FilterBankRtVehiclePositions.ParquetFilter.heavy_rail,
@@ -117,7 +117,7 @@ def start_gtfs_rt_parquet_updates_local() -> None:
     HyperGtfsRtTripUpdatesHeavyRail = FilteredHyperJob(
         remote_input_location=springboard_rt_trip_updates,
         remote_output_location=tableau_rt_trip_updates_heavyrail_30_day,
-        start_date=days_ago(30),
+        start_date=30,
         processed_schema=convert_gtfs_rt_trip_updates.HeavyRailTerminalTripUpdates.to_pyarrow_schema(),
         dataframe_filter=convert_gtfs_rt_trip_updates.heavyrail,
         parquet_filter=FilterBankRtTripUpdates.ParquetFilter.heavy_rail,
@@ -127,7 +127,7 @@ def start_gtfs_rt_parquet_updates_local() -> None:
     HyperDevGreenGtfsRtVehiclePositions = FilteredHyperJob(
         remote_input_location=springboard_devgreen_rt_vehicle_positions,
         remote_output_location=tableau_devgreen_rt_vehicle_positions_lightrail_60_day,
-        start_date=days_ago(60),
+        start_date=60,
         processed_schema=convert_gtfs_rt_vehicle_position.LightRailTerminalVehiclePositions.to_pyarrow_schema(),
         dataframe_filter=convert_gtfs_rt_vehicle_position.lrtp,
         parquet_filter=FilterBankRtVehiclePositions.ParquetFilter.light_rail,
@@ -137,7 +137,7 @@ def start_gtfs_rt_parquet_updates_local() -> None:
     HyperDevGreenGtfsRtTripUpdates = FilteredHyperJob(
         remote_input_location=springboard_devgreen_rt_trip_updates,
         remote_output_location=tableau_devgreen_rt_trip_updates_lightrail_60_day,
-        start_date=days_ago(60),
+        start_date=60,
         processed_schema=convert_gtfs_rt_trip_updates.LightRailTerminalTripUpdates.to_pyarrow_schema(),
         dataframe_filter=convert_gtfs_rt_trip_updates.lrtp_devgreen,
         parquet_filter=FilterBankRtTripUpdates.ParquetFilter.light_rail,

runners/run_parquet_hyper_lamp.py

@@ -4,9 +4,8 @@
 
 from lamp_py.tableau.hyper import HyperJob
 from lamp_py.tableau.jobs.bus_performance import HyperBusPerformanceRecent, HyperBusPerformanceAll
-from lamp_py.tableau.jobs.filtered_hyper import FilteredHyperJob, days_ago
-from datetime import date
 from lamp_py.tableau.jobs.filtered_hyper import FilteredHyperJob
+from datetime import date, timedelta
 
 from lamp_py.runtime_utils.remote_files import (
     LAMP,
@@ -47,18 +46,18 @@
 TestHyperBusPerformanceRecent = HyperBusPerformanceRecent
 TestHyperBusPerformanceAll = HyperBusPerformanceAll
 
-
-TestHyperGtfsRtVehiclePositions.start_date = days_ago(1)
-TestHyperGtfsRtTripUpdates.start_date = days_ago(1)
-TestHyperGtfsRtVehiclePositionsHeavyRail.start_date = days_ago(1)
-TestHyperGtfsRtTripUpdatesHeavyRail.start_date = days_ago(1)
-TestHyperGtfsRtVehiclePositionsAllLightRail.start_date = days_ago(1)
-TestHyperDevGreenGtfsRtVehiclePositions.start_date = days_ago(1)
-TestHyperDevGreenGtfsRtTripUpdates.start_date = days_ago(1)
-TestHyperBusOperatorMappingRecent.start_date = days_ago(1)
-TestHyperBusOperatorMappingAll.start_date = days_ago(1)
-# TestHyperRtRailSubway.start_date=days_ago(1)
-# TestHyperRtRailCommuter.start_date=days_ago(1)
+yesterday = date.today() - timedelta(1)
+TestHyperGtfsRtVehiclePositions.start_date = yesterday
+TestHyperGtfsRtTripUpdates.start_date = yesterday
+TestHyperGtfsRtVehiclePositionsHeavyRail.start_date = yesterday
+TestHyperGtfsRtTripUpdatesHeavyRail.start_date = yesterday
+TestHyperGtfsRtVehiclePositionsAllLightRail.start_date = yesterday
+TestHyperDevGreenGtfsRtVehiclePositions.start_date = yesterday
+TestHyperDevGreenGtfsRtTripUpdates.start_date = yesterday
+TestHyperBusOperatorMappingRecent.start_date = yesterday
+TestHyperBusOperatorMappingAll.start_date = yesterday
+# TestHyperRtRailSubway.start_date=yesterday
+# TestHyperRtRailCommuter.start_date=yesterday
 
 TestHyperBus = FilteredHyperJob(
     remote_input_location=bus_events,

src/lamp_py/tableau/jobs/filtered_hyper.py

@@ -1,5 +1,5 @@
 from typing import Callable
-from datetime import date, datetime, timedelta
+from datetime import date, timedelta
 import pyarrow
 import pyarrow.parquet as pq
 import pyarrow.dataset as pd
@@ -16,11 +16,6 @@
 from lamp_py.aws.s3 import file_list_from_s3, file_list_from_s3_date_range
 
 
-def days_ago(num_days: int) -> date:
-    """helper function to get a date() object set to num_days ago"""
-    return (datetime.now() - timedelta(days=num_days)).date()
-
-
 # pylint: disable=R0917,R0902,R0913
 # pylint too many local variables (more than 15)
 class FilteredHyperJob(HyperJob):
@@ -32,20 +27,27 @@ def __init__(
         remote_output_location: S3Location,
         processed_schema: pyarrow.schema,
         tableau_project_name: str,
-        start_date: date | None = days_ago(7),  # default this the past week of data
-        end_date: date | None = datetime.now().date(),
+        start_date: date | int | None = None,
+        end_date: date | None = None,
         partition_template: str = "year={yy}/month={mm}/day={dd}/",
         parquet_preprocess: Callable[[pyarrow.Table], pyarrow.Table] | None = None,
         parquet_filter: pc.Expression | None = None,
         dataframe_filter: Callable[[pl.DataFrame], pl.DataFrame] | None = None,
     ) -> None:
+        """Validate start_date and end_date and assign properties if so."""
         HyperJob.__init__(
             self,
             hyper_file_name=remote_output_location.prefix.rsplit("/")[-1].replace(".parquet", ".hyper"),
             remote_parquet_path=remote_output_location.s3_uri,
             lamp_version=remote_output_location.version,
             project_name=tableau_project_name,
         )
+        if end_date is not None:
+            assert isinstance(start_date, date)
+        elif isinstance(start_date, int):
+            end_date = date.today()
+            start_date = end_date - timedelta(days=start_date)
+
         self.remote_input_location = remote_input_location
         self.remote_output_location = remote_output_location
         self.processed_schema = processed_schema  # expected output schema passed in

src/lamp_py/tableau/jobs/lamp_jobs.py

@@ -37,7 +37,7 @@
 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.filtered_hyper import FilteredHyperJob
 from lamp_py.tableau.jobs.rt_rail import HyperRtRail
 from lamp_py.utils.filter_bank import FilterBankRtTripUpdates, FilterBankRtVehiclePositions
 
@@ -47,7 +47,7 @@
 HyperGtfsRtVehiclePositions = FilteredHyperJob(
     remote_input_location=springboard_rt_vehicle_positions,
     remote_output_location=tableau_rt_vehicle_positions_lightrail_60_day,
-    start_date=days_ago(60),
+    start_date=60,
     processed_schema=convert_gtfs_rt_vehicle_position.LightRailTerminalVehiclePositions.to_pyarrow_schema(),
     dataframe_filter=convert_gtfs_rt_vehicle_position.lrtp,
     parquet_filter=FilterBankRtVehiclePositions.ParquetFilter.light_rail,
@@ -57,7 +57,7 @@
 HyperGtfsRtTripUpdates = FilteredHyperJob(
     remote_input_location=springboard_lrtp_trip_updates,
     remote_output_location=tableau_rt_trip_updates_lightrail_60_day,
-    start_date=days_ago(60),
+    start_date=60,
     processed_schema=convert_gtfs_rt_trip_updates.LightRailTerminalTripUpdates.to_pyarrow_schema(),
     dataframe_filter=convert_gtfs_rt_trip_updates.lrtp_prod,
     parquet_filter=FilterBankRtTripUpdates.ParquetFilter.light_rail,
@@ -67,7 +67,7 @@
 HyperGtfsRtVehiclePositionsHeavyRail = FilteredHyperJob(
     remote_input_location=springboard_rt_vehicle_positions,
     remote_output_location=tableau_rt_vehicle_positions_heavyrail_30_day,
-    start_date=days_ago(30),
+    start_date=30,
     processed_schema=convert_gtfs_rt_vehicle_position.HeavyRailTerminalVehiclePositions.to_pyarrow_schema(),
     dataframe_filter=convert_gtfs_rt_vehicle_position.heavyrail,
     parquet_filter=FilterBankRtVehiclePositions.ParquetFilter.heavy_rail,
@@ -77,7 +77,7 @@
 HyperGtfsRtTripUpdatesHeavyRail = FilteredHyperJob(
     remote_input_location=springboard_rt_trip_updates,
     remote_output_location=tableau_rt_trip_updates_heavyrail_30_day,
-    start_date=days_ago(30),
+    start_date=30,
     processed_schema=convert_gtfs_rt_trip_updates.HeavyRailTerminalTripUpdates.to_pyarrow_schema(),
     dataframe_filter=convert_gtfs_rt_trip_updates.heavyrail,
     parquet_filter=FilterBankRtTripUpdates.ParquetFilter.heavy_rail,
@@ -87,7 +87,7 @@
 HyperGtfsRtVehiclePositionsAllLightRail = FilteredHyperJob(
     remote_input_location=springboard_rt_vehicle_positions,
     remote_output_location=tableau_rt_vehicle_positions_all_light_rail_7_day,
-    start_date=days_ago(7),
+    start_date=7,
     processed_schema=convert_gtfs_rt_vehicle_position.VehiclePositions.to_pyarrow_schema(),
     dataframe_filter=convert_gtfs_rt_vehicle_position.apply_gtfs_rt_vehicle_positions_timezone_conversions,
     parquet_filter=FilterBankRtVehiclePositions.ParquetFilter.light_rail,
@@ -97,7 +97,7 @@
 HyperDevGreenGtfsRtVehiclePositions = FilteredHyperJob(
     remote_input_location=springboard_devgreen_rt_vehicle_positions,
     remote_output_location=tableau_devgreen_rt_vehicle_positions_lightrail_60_day,
-    start_date=days_ago(60),
+    start_date=60,
     processed_schema=convert_gtfs_rt_vehicle_position.LightRailTerminalVehiclePositions.to_pyarrow_schema(),
     dataframe_filter=convert_gtfs_rt_vehicle_position.lrtp,
     parquet_filter=FilterBankRtVehiclePositions.ParquetFilter.light_rail,
@@ -107,7 +107,7 @@
 HyperDevGreenGtfsRtTripUpdates = FilteredHyperJob(
     remote_input_location=springboard_devgreen_lrtp_trip_updates,
     remote_output_location=tableau_devgreen_rt_trip_updates_lightrail_60_day,
-    start_date=days_ago(60),
+    start_date=60,
     processed_schema=convert_gtfs_rt_trip_updates.LightRailTerminalTripUpdates.to_pyarrow_schema(),
     dataframe_filter=convert_gtfs_rt_trip_updates.lrtp_devgreen,
     parquet_filter=FilterBankRtTripUpdates.ParquetFilter.light_rail,
@@ -117,7 +117,7 @@
 HyperBusOperatorMappingRecent = FilteredHyperJob(
     remote_input_location=bus_operator_mapping,
     remote_output_location=tableau_bus_operator_mapping_recent,
-    start_date=days_ago(7),
+    start_date=7,
     processed_schema=TMDailyWorkPiece.to_pyarrow_schema(),
     dataframe_filter=None,
     parquet_filter=None,
@@ -128,7 +128,7 @@
 HyperBusOperatorMappingAll = FilteredHyperJob(
     remote_input_location=bus_operator_mapping,
     remote_output_location=tableau_bus_operator_mapping_all,
-    start_date=days_ago(60),
+    start_date=60,
     processed_schema=TMDailyWorkPiece.to_pyarrow_schema(),
     dataframe_filter=None,
     parquet_filter=None,