NWS/NOAA weather alert fetcher derived from https://github.com/greencoder/noaa-alerts-pushover

weather_alerts.py

import time
import lxml.etree
import requests
# import dateutil.parser as dateparser
import cachetools
from time import strftime
from apscheduler.schedulers.background import BackgroundScheduler


ATOM_NS = "{http://www.w3.org/2005/Atom}"
CAP_NS = "{urn:oasis:names:tc:emergency:cap:1.1}"
NWS_FEED_URL = "https://alerts.weather.gov/cap/us.php?x=0"
# NWS_FEED_URL = 'http://alerts.weather.gov/cap/or.php?x=0' # OR only


class NOAAAlert(object):
    """ This holds the stuff we need to care about in a NWS/NOAA alert """
    def __init__(
            self, id, title, event, details, expires, link, fips_codes,
            ugc_codes):
        self._id = id
        self._title = title
        self._event = event
        self._details = details
        self._expires = expires
        self._link = link
        self._fips_codes = fips_codes
        self._ugc_codes = ugc_codes
        self._checked = False  # have we examined this alert for notifications
        self._notified = False  # have we sent a notification about this alert

    @property
    def id(self):
        return self._id

    @property
    def title(self):
        return self._title

    @property
    def details(self):
        return self._details

    @property
    def expires(self):
        return self._expires

    @property
    def link(self):
        return self._link

    @property
    def checked(self):
        return self._checked

    @checked.setter
    def checked(self, value):
        self._checked = value

    @property
    def notified(self):
        return self._notified

    @notified.setter
    def notified(self, value):
        self._notified = value

    def __repr__(self):
        return "id: {0}\ntitle {1}\nevent: {2}\ndetails: {3}\n\
expires: {4}\nlink: {5}\nlink: {6}\nfips: {7}\nugc: {8}".format(
            self._id, self._title, self._event, self._details, self._link,
            self._expires, self._link, self._fips_codes, self._ugc_codes)


class AlertFetcher(object):
    """ Fetch some alerts from NOAA"""

    def __init__(self):
        # TODO populate these from chats
        self._fips_codes = ['041005', '041051', '041067', '054103']
        self._ugc_codes = []
        self.alert_cache = cachetools.TTLCache(maxsize=1000, ttl=600)
        self.scheduler = BackgroundScheduler()
        self.scheduler.start()
        self._no_alerts = False

    def fetch_feed(self):
        """ Fetch NOAA alerts xml feed """
        response = requests.get(NWS_FEED_URL)
        tree = lxml.etree.fromstring(response.text.encode('utf-8'))

        # TODO check if there are more of these that we should care about
        severe_special_types = ['Thunderstorm',
                                'Strong Storm',
                                'Wind', 'Rain',
                                'Hail',
                                'Tornado',
                                'Flood']
        # TODO this list isn't required because the cache gets populated later
        alerts = []

        for entry_el in tree.findall(ATOM_NS + 'entry'):
            entry_id = entry_el.find(ATOM_NS + 'id').text
            title = entry_el.find(ATOM_NS + 'title').text
            if title == "There are no active watches, warnings or advisories":
                if not self._no_alerts:
                    # TODO notify about 'no active alerts'
                    print "There are no active alerts at {0}".format(
                        strftime("%Y-%m-%d %H:%M"))
                    self._no_alerts = True
                return
            self._no_alerts = False
            if entry_el.find(CAP_NS + 'event') is not None:
                event = entry_el.find(CAP_NS + 'event').text
            else:
                event = None
            expires_text = entry_el.find(CAP_NS + 'expires').text
            expires_text
            url = entry_el.find(ATOM_NS + 'link').attrib['href']
            fips_list, ugc_list = self.parse_geocode_from_entry(entry_el)
            special_severe = []
            if event in ('Severe Weather Statement',
                         'Special Weather Statement'):
                summary = entry_el.find(ATOM_NS + 'summary').text.upper()
                for item in severe_special_types:
                    if item.upper() in summary:
                        special_severe.append(item)

            alert = NOAAAlert(
                entry_id, title, event, ', '.join(
                    special_severe), expires_text, url, fips_list, ugc_list)
            alerts.append(alert)
            # TODO to give this entry a TTL of (expiry - now) seconds
            self.alert_cache[entry_id] = alert

        # print "There were {0} alerts at {1}".format(
        #    len(alerts), strftime("%Y-%m-%d %H:%M"))
        # print "{0} alerts in cache".format(len(self.alert_cache))
        self.check_alerts()
        # print "checked alerts"

    def parse_geocode_from_entry(self, entry):
        """ Parse the FIPS6 and UGC geocodes from a NOAA feed entry """
        geocode_element = entry.find(CAP_NS + 'geocode')
        fips_list = []
        ugc_list = []

        if geocode_element is not None:
            for value_name_element in geocode_element.findall(
                    ATOM_NS + 'valueName'):
                if value_name_element.text == "FIPS6":
                    fips_element = value_name_element.getnext()
                    if (fips_element is not None
                            and fips_element.text is not None):
                        fips_list = fips_element.text.split(' ')
                elif value_name_element.text == 'UGC':
                    ugc_element = value_name_element.getnext()
                    if (ugc_element is not None
                            and ugc_element.text is not None):
                        ugc_list = ugc_element.text.split(' ')
        return fips_list, ugc_list

    def fetch_details(self, alert):
        """ Fetch details for a specific alert """
        # TODO maybe prettify this
        print "fetching details from  {0}".format(alert.link)
        # detail_response = requests.get(alert.link)
        # print detail_response.text.encode('utf-8')

    def check_alerts(self):
        for alert_id in self.alert_cache:
            alert = self.alert_cache[alert_id]
            print "checking {0}".format(alert.id)
            if alert.checked:
                pass
            else:
                fips_match = set(
                    alert._fips_codes).intersection(self._fips_codes)
                alert.checked = True
                # print "matched {0} fips codes".format(len(fips_match))
                if len(fips_match) > 0:
                    # print "matched {0} for {1}".format(
                    #    alert.title, fips_match)
                    if not alert.notified:
                        # send notification
                        self.notify_about_alert(alert)
                        alert.notified = True

    def notify_about_alert(self, alert):
        print "totally notifying about {0}".format(alert.id)


if __name__ == '__main__':
    fetcher = AlertFetcher()
    fetcher.fetch_feed()
    fetcher.scheduler.add_job(fetcher.fetch_feed, 'interval', minutes=2)
    try:
        while True:
            time.sleep(2)
    except (KeyboardInterrupt, SystemExit):
        fetcher.scheduler.shutdown()