Skip to content

Instantly share code, notes, and snippets.

@bdunnette
Created August 11, 2023 18:57
Show Gist options
  • Save bdunnette/7e308bb96b44cae8c8175ce667fae9d0 to your computer and use it in GitHub Desktop.
Save bdunnette/7e308bb96b44cae8c8175ce667fae9d0 to your computer and use it in GitHub Desktop.
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 = "brian@dunnette.us"
DEFAULT_RECIPIENTS = ["bdunnette@gmail.com", "brian@dunnette.us"]
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)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment