ClearHour: telling every Delhi school its cleanest hour, on AWS

Events:
  Morning:
    Type: ScheduleV2
    Properties:
      ScheduleExpression: cron(30 5 * * ? *)
      ScheduleExpressionTimezone: Asia/Kolkata
      Input: '{"source": "live"}'

"SendAlerts": {
  "Type": "Map",
  "ItemsPath": "$.alerts",
  "MaxConcurrency": 5,
  "ItemProcessor": { "ProcessorConfig": { "Mode": "INLINE" }, "StartAt": "Send", "States": { "...": "..." } },
  "ResultPath": "$.sent",
  "End": true
}

FROM public.ecr.aws/lambda/python:3.13
# LightGBM needs the OpenMP runtime; the base image is Amazon Linux 2023 minimal
RUN dnf install -y libgomp && dnf clean all
COPY functions/forecast/requirements.txt ${LAMBDA_TASK_ROOT}/
RUN pip install --no-cache-dir -r ${LAMBDA_TASK_ROOT}/requirements.txt
COPY src/clearhour ${LAMBDA_TASK_ROOT}/clearhour
CMD ["clearhour.handlers.forecast.handler"]

def move_alert(school_id, day, from_status, to_status, **attrs) -> bool:
    """Atomically moves an alert between statuses; False if it wasn't in from_status."""
    try:
        table().update_item(
            Key={"pk": f"SCHOOL#{school_id}", "sk": f"ALERT#{day}"},
            UpdateExpression="SET #s = :to" + "".join(f", #{k} = :{k}" for k in attrs),
            ConditionExpression="#s = :from",
            ExpressionAttributeNames={"#s": "status", **{f"#{k}": k for k in attrs}},
            ExpressionAttributeValues={":from": from_status, ":to": to_status, **{f":{k}": v for k, v in attrs.items()}},
        )
        return True
    except ClientError as e:
        if e.response["Error"]["Code"] == "ConditionalCheckFailedException":
            return False
        raise

# EUM rejects bare digits ("Invalid destination phone number"); WhatsApp sends, and we store, no "+"
to = str(payload.get("to", ""))
payload = {**payload, "to": to if to.startswith("+") else "+" + to}
resp = boto3.client("socialmessaging").send_whatsapp_message(
    originationPhoneNumberId=os.environ["WA_PHONE_NUMBER_ID"],
    message=json.dumps(payload).encode(),
    metaApiVersion=version,
)

_s3 = boto3.client(
    "s3",
    region_name="us-east-1",
    config=Config(signature_version=UNSIGNED, retries={"max_attempts": 10, "mode": "adaptive"}),
)

def list_month_keys(location_id: int, year: int, month: int) -> list[str]:
    """Every daily file for one location-month. Lists the prefix rather than guessing file names."""
    keys, kwargs = [], {"Bucket": "openaq-data-archive",
                        "Prefix": f"records/csv.gz/locationid={location_id}/year={year}/month={month:02d}/"}
    while True:
        resp = _s3.list_objects_v2(**kwargs)
        keys += [o["Key"] for o in resp.get("Contents", []) if o["Key"].endswith(".csv.gz")]
        if not resp.get("IsTruncated"):
            return keys
        kwargs["ContinuationToken"] = resp["NextContinuationToken"]