nws_notifier.py

from datetime import datetime
import logging
import requests

logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s")

# Start Configuration
NWS_STATION = "KMIC"
MIN_PRECIP = 10

DEFAULT_SENDER = "[email protected]"
DEFAULT_RECIPIENTS = ["[email protected]", "[email protected]"]

MAILGUN_URL = "https://api.mailgun.net/v3/sandbox1c589e2a0d85487d9550d5fac026c1a6.mailgun.org/messages"
MAILGUN_KEY = "this-is-not-a-real-key"
# End Configuration


def get_nws_obs(station="KMIC"):
    """Get the latest observation from the NWS API

    Args:
        station (str, optional): NWS station code to retrieve observations. Defaults to "KMIC".

    Returns:
        tuple: (datetime, float) - observation time and precipitation in the last 6 hours
    """
    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(
    sender=DEFAULT_SENDER,
    to=DEFAULT_RECIPIENTS,
    subject="Test Email",
    text="TEST EMAIL - PLEASE IGNORE",
):
    """Send a message via Mailgun

    Args:
        sender (str, optional): Sender email address. Defaults to DEFAULT_SENDER.
        to (list, optional): List of recipient email addresses. Defaults to DEFAULT_RECIPIENTS.
        subject (str, optional): Subject line of email. Defaults to "Test Email".
        text (str, optional): Body of email. Defaults to "TEST EMAIL - PLEASE IGNORE".

    Returns:
        requests.Response: Response from Mailgun API
    """
    try:
        sent = requests.post(
            MAILGUN_URL,
            auth=("api", MAILGUN_KEY),
            data={"from": sender, "to": to, "subject": subject, "text": text},
        )
        sent.raise_for_status()
    except requests.exceptions.HTTPError as e:
        logger.error(e)


if __name__ == "__main__":
    obs_time, precipitationLast6Hours = get_nws_obs(station=NWS_STATION)
    time_formatted = obs_time.strftime("%a %b %d %H:%M %Z")
    if precipitationLast6Hours == None or precipitationLast6Hours < MIN_PRECIP:
        logger.info(
            f"As of {time_formatted}, precipitation in the last 6 hours at station {NWS_STATION} was {precipitationLast6Hours} (<{MIN_PRECIP}) - not sending notifications"
        )
    else:
        msg = f"As of {time_formatted}, there had been {precipitationLast6Hours}mm of precipitation in the last 6 hours at station {NWS_STATION} - please perform a SWPP inspection at your earliest convenience."
        logger.info(msg)
        sent = send_message(
            text=msg, subject=f"SWPP Notification - {NWS_STATION} - {time_formatted}"
        )
        logger.debug(sent.content)