Update for Daily Delay Tracking

* Daily Delay Updates

Main Update:
Added support for new daily delays stored in a new dynamo table. Weekly updates now call data from the new daily dynamo table for aggregation.

* QoL and cloud config updates to daily delay tracking

Fixed PR Comments:
- Removed authorship comment
- Stores initialized delay type dict as a constant for accessing in process.py (reducing duplicate code)
- Delayed weekly delay update by 15 minutes to allow daily update to run
- Cloud config support added in Chalice json files

Author: Icebox00

ingestor/.chalice/config.json

@@ -106,7 +106,12 @@
           "lambda_memory_size": 256,
           "lambda_timeout": 90
         },
-        "update_alert_delays": {
+        "update_weekly_alert_delays": {
+          "iam_policy_file": "policy-alert-delays.json",
+          "lambda_memory_size": 256,
+          "lambda_timeout": 300
+        },
+        "update_daily_alert_delays": {
           "iam_policy_file": "policy-alert-delays.json",
           "lambda_memory_size": 256,
           "lambda_timeout": 300

ingestor/.chalice/policy-alert-delays.json

@@ -14,7 +14,15 @@
       "Effect": "Allow",
       "Action": ["dynamodb:BatchWriteItem"],
       "Resource": [
-        "arn:aws:dynamodb:us-east-1:473352343756:table/AlertDelaysWeekly"
+        "arn:aws:dynamodb:us-east-1:473352343756:table/AlertDelaysWeekly",
+        "arn:aws:dynamodb:us-east-1:473352343756:table/AlertDelaysDaily"
+      ]
+    },
+    {
+      "Effect": "Allow",
+      "Action": ["dynamodb:Query"],
+      "Resource": [
+        "arn:aws:dynamodb:us-east-1:473352343756:table/AlertDelaysDaily"
       ]
     }
   ]

ingestor/.chalice/resources.json

@@ -199,7 +199,28 @@
         }
       }
     },
-    "UpdateAlertDelays": {
+    "UpdateWeeklyAlertDelays": {
+      "Type": "AWS::Serverless::Function",
+      "Properties": {
+        "Environment": {
+          "Variables": {
+            "DD_API_KEY": {
+              "Ref": "DDApiKey"
+            },
+            "DD_VERSION": {
+              "Ref": "GitVersion"
+            },
+            "DD_TAGS": {
+              "Ref": "DDTags"
+            },
+            "DD_GIT_REPOSITORY_URL": {
+              "Ref": "DDGitRepositoryUrl"
+            }
+          }
+        }
+      }
+    },
+    "UpdateDailyAlertDelays": {
       "Type": "AWS::Serverless::Function",
       "Properties": {
         "Environment": {
@@ -488,6 +509,39 @@
         ]
       }
     },
+    "AlertDelaysDailyDB": {
+      "Type": "AWS::DynamoDB::Table",
+      "Properties": {
+        "TableName": "AlertDelaysDaily",
+        "KeySchema": [
+          {
+            "AttributeName": "line",
+            "KeyType": "HASH"
+          },
+          {
+            "AttributeName": "date",
+            "KeyType": "RANGE"
+          }
+        ],
+        "AttributeDefinitions": [
+          {
+            "AttributeName": "line",
+            "AttributeType": "S"
+          },
+          {
+            "AttributeName": "date",
+            "AttributeType": "S"
+          }
+        ],
+        "BillingMode": "PAY_PER_REQUEST",
+        "Tags": [
+          {
+            "Key": "service",
+            "Value": "t-performance-dash"
+          }
+        ]
+      }
+    },
     "DeliveredTripMetricsMonthlyDB": {
       "Type": "AWS::DynamoDB::Table",
       "Properties": {

ingestor/app.py

@@ -122,13 +122,22 @@ def update_trip_metrics(event):
     trip_metrics.ingest_recent_trip_metrics(lookback_days=7)
 
 
-# 8:30am UTC -> 3:30/4:30am ET every Monday and Tuesday
+# 8:30am UTC -> 3:45/4:45am ET every Monday and Tuesday
 # There's shouldn't be any benefit to running it more frequently.
-@app.schedule(Cron(30, 8, "?", "*", "MON,TUE", "*"))
-def update_alert_delays(event):
+# Runs 15 minutes after the daily update
+@app.schedule(Cron(45, 8, "?", "*", "MON,TUE", "*"))
+def update_weekly_alert_delays(event):
     today = datetime.now()
     one_week_ago = (today - timedelta(days=15)).date()
-    delays.update_table(one_week_ago, today.date())
+    delays.update_weekly_from_daily(one_week_ago, today.date())
+
+
+# for daily delay uploads
+@app.schedule(Cron(30, 8, "*", "*", "?", "*"))
+def update_daily_alert_delays(event):
+    today = datetime.now()
+    yesterday = (today - timedelta(days=1)).date()
+    delays.update_table(yesterday, today.date())
 
 
 # Manually triggered lambda for populating daily trip metric tables. Only needs to be ran once.

ingestor/chalicelib/constants.py

@@ -338,3 +338,22 @@ def get_weekly_table_update_start():
     "line-orange": ["line-orange"],
     "line-mattapan": ["line-mattapan"],
 }
+
+# Initializing all Delay types at 0 for processing
+DELAY_BY_TYPE = {
+    "disabled_vehicle": 0,
+    "signal_problem": 0,
+    "power_problem": 0,
+    "door_problem": 0,
+    "brake_problem": 0,
+    "switch_problem": 0,
+    "track_issue": 0,
+    "mechanical_problem": 0,
+    "track_work": 0,
+    "car_traffic": 0,
+    "police_activity": 0,
+    "medical_emergency": 0,
+    "fire": 0,
+    "flooding": 0,
+    "other": 0,
+}

ingestor/chalicelib/delays/__init__.py

@@ -1,3 +1,3 @@
-__all__ = ["update_table"]
+__all__ = ["update_table", "update_weekly_from_daily"]
 
-from .process import update_table
+from .process import update_table, update_weekly_from_daily

ingestor/chalicelib/delays/aggregate.py

@@ -40,10 +40,50 @@ def group_weekly_data(df: pd.DataFrame, start_date: str):
     df_weekly = df_weekly.fillna(0)
     # Pandas resample uses the end date of the range as the index. So we subtract 6 days to convert to first date of the range.
     df_weekly.index = df_weekly.index - pd.Timedelta(days=6)
-    # Drop the first week if it is incomplete
-    if datetime.fromisoformat(start_date).weekday() != 0:
-        df_weekly = df_weekly.tail(-1)
+
     # Convert date back to string.
     df_weekly["date"] = df_weekly.index.strftime("%Y-%m-%d")
     df_weekly = df_weekly[df_weekly["line"] != 0]
     return df_weekly.to_dict(orient="records")
+
+
+# Added by Icebox00 (temporary note)
+def group_daily_data(df: pd.DataFrame, start_date: str):
+    # Formats daily data on delays for Dynamo
+
+    df_daily = df.resample("D").agg(
+        {
+            "total_delay_time": np.sum,
+            "disabled_vehicle": np.sum,
+            "signal_problem": np.sum,
+            "power_problem": np.sum,
+            "brake_problem": np.sum,
+            "door_problem": np.sum,
+            "switch_problem": np.sum,
+            "track_issue": np.sum,
+            "track_work": np.sum,
+            "car_traffic": np.sum,
+            "mechanical_problem": np.sum,
+            "police_activity": np.sum,
+            "medical_emergency": np.sum,
+            "flooding": np.sum,
+            "fire": np.sum,
+            "other": np.sum,
+            "line": "min",
+        }
+    )
+
+    # Fill the empties with 0
+    df_daily = df_daily.fillna(0)
+
+    # Convert date back to string.
+    df_daily["date"] = df_daily.index.strftime("%Y-%m-%d")
+
+    df_daily = df_daily.reset_index(drop=True)
+
+    df_daily = df_daily[df_daily["line"] != 0]
+
+    # Select only the columns we need for storage
+    daily_records = df_daily.to_dict(orient="records")
+
+    return daily_records

ingestor/chalicelib/delays/process.py

@@ -1,6 +1,7 @@
 from datetime import date, timedelta
 from decimal import Decimal
 import json
+from boto3.dynamodb.conditions import Key
 import re
 from typing import List
 from urllib.parse import urlencode
@@ -9,10 +10,11 @@
 import requests
 
 from chalicelib import constants, dynamo
-from chalicelib.delays.aggregate import group_weekly_data
+from chalicelib.delays.aggregate import group_weekly_data, group_daily_data
 from chalicelib.delays.types import Alert, AlertsRequest
 
-TABLE_NAME = "AlertDelaysWeekly"
+WEEKLY_TABLE_NAME = "AlertDelaysWeekly"
+DAILY_TABLE_NAME = "AlertDelaysDaily"
 
 
 def generate_requests(start_date: date, end_date: date, lines=constants.ALL_LINES) -> List[AlertsRequest]:
@@ -193,23 +195,8 @@ def process_delay_time(alerts: List[Alert]):
                 }
             )
     total_delay = 0
-    delay_by_type = {
-        "disabled_vehicle": 0,
-        "signal_problem": 0,
-        "power_problem": 0,
-        "door_problem": 0,
-        "brake_problem": 0,
-        "switch_problem": 0,
-        "track_issue": 0,
-        "mechanical_problem": 0,
-        "track_work": 0,
-        "car_traffic": 0,
-        "police_activity": 0,
-        "medical_emergency": 0,
-        "fire": 0,
-        "flooding": 0,
-        "other": 0,
-    }
+    delay_by_type = constants.DELAY_BY_TYPE.copy()
+
     for delay in delays:
         if (delay is None) or (len(delay) == 0):
             continue
@@ -227,16 +214,21 @@ def process_requests(requests: List[AlertsRequest], lines=constants.ALL_LINES):
 
     for request in requests:
         data = process_single_day(request)
+        # Initializing at 0 regardless of condition
+        total_delay = 0
+        delay_by_type = constants.DELAY_BY_TYPE.copy()
+
         if data is not None and len(data) != 0:
             total_delay, delay_by_type = process_delay_time(data)
-            all_data[request.route].append(
-                {
-                    "date": request.date.isoformat(),
-                    "line": request.route,
-                    "total_delay_time": total_delay,
-                    "delay_by_type": delay_by_type,
-                }
-            )
+        # We should always append zero records just in case
+        all_data[request.route].append(
+            {
+                "date": request.date.isoformat(),
+                "line": request.route,
+                "total_delay_time": total_delay,
+                "delay_by_type": delay_by_type,
+            }
+        )
 
     df_data = {}
     for line in lines:
@@ -249,6 +241,55 @@ def process_requests(requests: List[AlertsRequest], lines=constants.ALL_LINES):
     return df_data
 
 
+def get_daily_data_for_week(start_date: date, end_date: date, lines=constants.ALL_LINES):
+    """
+    Query daily data from DynamoDB for weekly aggregation.
+    Done as opposed to performing another API call for weekly aggregation.
+    Also guarantees complete info for every day.
+    """
+    daily_records = []
+
+    for line in lines:
+        # Convert dates to ISO format strings to match DynamoDB storage format
+        start_date_str = start_date.isoformat()
+        end_date_str = end_date.isoformat()
+
+        # Query DynamoDB for this line and date range
+        params = {
+            "KeyConditionExpression": Key("line").eq(line)
+            & Key("date").between(start_date_str, end_date_str)  # Use string format
+        }
+        try:
+            line_data = dynamo.query_dynamo(params, DAILY_TABLE_NAME)
+            daily_records.extend(line_data)
+        except Exception as e:
+            print(f"Error querying {line}: {e}")
+
+    return daily_records
+
+
+def update_weekly_from_daily(start_date: date, end_date: date, lines=constants.ALL_LINES):
+    """
+    Update weekly table by aggregating daily data from DynamoDB.
+    Avoids another api call.
+    """
+
+    # Get daily data from DynamoDB instead of API
+    daily_records = get_daily_data_for_week(start_date, end_date, lines)
+
+    # Convert to DataFrame and process
+    df = pd.DataFrame(daily_records)
+    df["date"] = pd.to_datetime(df["date"])
+    df.set_index("date", inplace=True)
+
+    # Aggregate weekly for each line
+    weekly_data = []
+    for line, line_df in df.groupby("line"):
+        weekly_data.extend(group_weekly_data(line_df, start_date.isoformat()))
+
+    dynamo.dynamo_batch_write(json.loads(json.dumps(weekly_data), parse_float=Decimal), WEEKLY_TABLE_NAME)
+
+
 def update_table(start_date: date, end_date: date, lines=constants.ALL_LINES):
     """
     Update the table with rapid transit data
@@ -258,12 +299,13 @@ def update_table(start_date: date, end_date: date, lines=constants.ALL_LINES):
 
     grouped_data = []
     for line, df in all_data.items():
-        grouped_data.extend(group_weekly_data(df, start_date.isoformat()))
-
-    dynamo.dynamo_batch_write(json.loads(json.dumps(grouped_data), parse_float=Decimal), TABLE_NAME)
+        grouped_data.extend(group_daily_data(df, start_date.isoformat()))
+    dynamo.dynamo_batch_write(json.loads(json.dumps(grouped_data), parse_float=Decimal), DAILY_TABLE_NAME)
 
 
+# Testing daily updates. Using random dates. Feel free to change and uncomment as needed.
 if __name__ == "__main__":
-    start_date = date(2025, 6, 1)
-    end_date = date(2025, 8, 10)
-    update_table(start_date, end_date, constants.ALL_LINES)
+    start_date = date(2025, 9, 15)
+    end_date = date(2025, 9, 21)
+    # update_table(start_date, end_date, constants.ALL_LINES)
+    # update_weekly_from_daily(start_date, end_date, constants.ALL_LINES)