Send email notifications if 6-hour precipitation is above threshold

.env

SENDGRID_API_KEY=Paste key from https://app.sendgrid.com/settings/api_keys here
[email protected]
RECIPIENTS=${SENDER},[email protected]
NWS_STATION=KMIC
MIN_PRECIP=25

README.md

RainyDay

1. Download and unzip all files
2. Sign up for a Sendgrid account: https://signup.sendgrid.com/
3. Get an API key from https://app.sendgrid.com/settings/api_keys
4. Paste API key into the SENDGRID_API_KEY field of the .env file
5. Set other .env variables (particularly SENDER and RECIPIENTS, which can be a comma-separated list of emails)
6. Ensure required Python libraries are installed: python -m pip install python-dotenv sendgrid
7. Run script: python rainy_day.py

rainy_day.py

import os
from datetime import datetime
import logging
import requests
from sendgrid import SendGridAPIClient
from sendgrid.helpers.mail import Mail
from dotenv import load_dotenv

load_dotenv()

consoleHandler = logging.StreamHandler()
consoleHandler.setFormatter(logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s'))

logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
logger.addHandler(consoleHandler)

# Weather station code from https://forecast.weather.gov/stations.php
NWS_STATION = os.getenv("NWS_STATION",default="KMIC")

# Minimium precipitation to alert for - set this to 0 for testing
MIN_PRECIP = float(os.getenv("MIN_PRECIP",default=25))

# Find your API key at https://app.sendgrid.com/settings/api_keys
SENDGRID_API_KEY = os.getenv("SENDGRID_API_KEY")

SENDER = os.getenv("SENDER",default="[email protected]")
RECIPIENTS = os.getenv("RECIPIENTS",default=[SENDER])

def get_nws_obs(station=NWS_STATION):
    """Get the latest weather conditions for a given NWS station"""
    try:
        observation = requests.get(
            f"https://api.weather.gov/stations/{station}/observations/latest"
        )
        observation.raise_for_status()
    except requests.exceptions.HTTPError as e:
        logger.error(e)

    obs_data = observation.json()["properties"]
    obs_time = datetime.fromisoformat(obs_data["timestamp"])
    precipitationLast6Hours = obs_data["precipitationLast6Hours"]["value"]
    return obs_time, precipitationLast6Hours

def send_message(
    from_email=SENDER,
    to_emails=RECIPIENTS,
    # Default subject and text - we'll override these when the function is actually run!
    subject="TEST NOTIFICATION - PLEASE IGNORE",
    html_content="<em>Test email - please ignore</em>"
):
    """Send a message via the Sendgrid API"""

    try:
        message = Mail(
          from_email=from_email,
          to_emails=to_emails,
          subject=subject,
          html_content=html_content)
        sg = SendGridAPIClient(SENDGRID_API_KEY)
        response = sg.send(message)
        logger.debug(response)
    except Exception as e:
        logger.error(e.message)

def send_message_if_precip():
    """Send a message if precipitation is above threshold"""
    obs_time, precipitation = get_nws_obs(NWS_STATION)
    precipitation = 0 if precipitation == None else precipitation
    obs_time = obs_time.strftime("%Y-%m-%d %H:%M %Z")
    msg = f"{NWS_STATION} at {obs_time} - {precipitation} mm of rain in the last 6 hours."
    if precipitation >= MIN_PRECIP:
        subject = f"RAIN ALERT for {NWS_STATION}"
        html_content = f"<strong>RAIN ALERT</strong> for {msg}"
        logger.info(html_content)
        send_message(from_email=SENDER,to_emails=RECIPIENTS,subject=subject,html_content=html_content)
    else:
        logger.info(
            f"No rain alert for {msg}"
        )

if __name__=="__main__":
    send_message_if_precip()