xmet/lib/nexrad/storm.py

import re
import gzip
import csv
import datetime

from nexrad.coord import Coord, COORD_SYSTEM
from nexrad.radar import RADAR_RANGE

def time_from_str(time: str):
    size = len(time)

    if size <= 2:
        return (
            int(time) % 24,
            0
        )

    return (
        int(time[0:size-2]) % 24,
        int(time[size-2:])  % 60
    )

def timestamp_from_parts(tz: datetime.tzinfo, yearmonth: str, day: str, time: str) -> datetime.datetime:
    hour, minute = time_from_str(time)

    return datetime.datetime(
        tzinfo = tz,
        year   = int(yearmonth[0:4]),
        month  = int(yearmonth[4:6]),
        day    = int(day),
        hour   = hour,
        minute = minute
    ).astimezone(datetime.UTC)

TIMEZONES = {
    '-12': datetime.timezone(datetime.timedelta(hours=-12)),
    '-11': datetime.timezone(datetime.timedelta(hours=-11)),
    '-10': datetime.timezone(datetime.timedelta(hours=-10)),
    '-9':  datetime.timezone(datetime.timedelta(hours=-9)),
    '-8':  datetime.timezone(datetime.timedelta(hours=-8)),
    '-7':  datetime.timezone(datetime.timedelta(hours=-7)),
    '-6':  datetime.timezone(datetime.timedelta(hours=-6)),
    '-5':  datetime.timezone(datetime.timedelta(hours=-5)),
    '-4':  datetime.timezone(datetime.timedelta(hours=-4)),
    '-3':  datetime.timezone(datetime.timedelta(hours=-3)),
    '-2':  datetime.timezone(datetime.timedelta(hours=-2)),
    '-1':  datetime.timezone(datetime.timedelta(hours=-1)),
    '-0':  datetime.UTC,
    '+0':  datetime.UTC,
    '+1':  datetime.timezone(datetime.timedelta(hours=1)),
    '+2':  datetime.timezone(datetime.timedelta(hours=2)),
    '+3':  datetime.timezone(datetime.timedelta(hours=3)),
    '+4':  datetime.timezone(datetime.timedelta(hours=4)),
    '+5':  datetime.timezone(datetime.timedelta(hours=5)),
    '+6':  datetime.timezone(datetime.timedelta(hours=6)),
    '+7':  datetime.timezone(datetime.timedelta(hours=7)),
    '+8':  datetime.timezone(datetime.timedelta(hours=8)),
    '+9':  datetime.timezone(datetime.timedelta(hours=9)),
    '+10': datetime.timezone(datetime.timedelta(hours=10)),
    '+11': datetime.timezone(datetime.timedelta(hours=11)),
    '+12': datetime.timezone(datetime.timedelta(hours=12)),
}

TIMEZONE_RE = re.compile(r'^(?:[A-Z]+)([+\-]\d)$')

def timezone_from_str(text: str) -> datetime.timezone:
    match = TIMEZONE_RE.match(text)

    if match is None:
        return datetime.UTC

    tz = TIMEZONES.get(match[1])

    return datetime.UTC if tz is None else tz

def coord_from_str(text_lon: str, text_lat: str):
    lon = 0.0 if text_lon == '' else float(text_lon)
    lat = 0.0 if text_lat == '' else float(text_lat)

    return Coord(lon, lat)

class StormReport():
    __slots__ = (
        'timestamp_start', 'timestamp_end', 'episode_id', 'event_id',
        'state', 'event_type', 'wfo', 'coord_start', 'coord_end',
        'locale_start', 'locale_end', 'tornado_f_rating'
    )

    timestamp_start:  datetime.datetime
    timestamp_end:    datetime.datetime
    episode_id:       int
    event_id:         int
    state:            str
    event_type:       str
    wfo:              str
    coord_start:      Coord
    coord_end:        Coord
    locale_start:     str
    locale_end:       str
    tornado_f_rating: str

    @staticmethod
    def from_csv_row(row: dict):
        report = StormReport()

        tz = timezone_from_str(row['CZ_TIMEZONE'])

        report.timestamp_start  = timestamp_from_parts(tz, row['BEGIN_YEARMONTH'], row['BEGIN_DAY'], row['BEGIN_TIME'])
        report.timestamp_end    = timestamp_from_parts(tz, row['END_YEARMONTH'],   row['END_DAY'],   row['END_TIME'])
        report.state            = row['STATE']
        report.event_type       = row['EVENT_TYPE']
        report.wfo              = row['WFO']
        report.coord_start      = coord_from_str(row['BEGIN_LON'], row['BEGIN_LAT'])
        report.coord_end        = coord_from_str(row['END_LON'],   row['END_LAT'])
        report.locale_start     = row['BEGIN_LOCATION']
        report.locale_end       = row['END_LOCATION']
        report.tornado_f_rating = row['TOR_F_SCALE']

        try:
            report.episode_id = int(row['EPISODE_ID'])
        except ValueError:
            report.episode_id = None

        try:
            report.event_id = int(row['EVENT_ID'])
        except ValueError:
            report.event_id = None

        return report

    @staticmethod
    def each_from_csv_file(file: str):
        with gzip.open(file, 'rt') as fh:
            reader = csv.DictReader(fh, dialect='excel')

            for row in reader:
                try:
                    yield StormReport.from_csv_row(row)
                except:
                    pass

    RADAR_SIGNIFICANT_EVENT_TYPES = {
        'Blizzard':                 True,
        'Coastal Flood':            True,
        'Debris Flow':              True,
        'Dust Storm':               True,
        'Flash Flood':              True,
        'Flood':                    True,
        'Funnel Cloud':             True,
        'Hail':                     True,
        'Heavy Rain':               True,
        'Heavy Snow':               True,
        'Hurricane (Typhoon)':      True,
        'Ice Storm':                True,
        'Lake-Effect Snow':         True,
        'Lightning':                True,
        'Marine Hail':              True,
        'Marine Strong Wind':       True,
        'Marine Thunderstorm Wind': True,
        'Seiche':                   True,
        'Storm Surge/Tide':         True,
        'Thunderstorm Wind':        True,
        'Tornado':                  True,
        'Tropical Depression':      True,
        'Tropical Storm':           True,
        'Waterspout':               True,
        'Winter Storm':             True,
    }

    def is_radar_significant(self):
        return self.event_type in self.RADAR_SIGNIFICANT_EVENT_TYPES

    def nearby_radars(self, db):
        sql = """
            select
                id,
                call,
                ST_Distance(
                    coord,
                    MakeLine(MakePoint(?, ?, {csr}),
                             MakePoint(?, ?, {csr})),
                    true) as distance
            from
                nexrad_radar
            where
                distance <= ?
            order by
                distance asc
        """.format(csr=COORD_SYSTEM)

        radars = list()

        st = db.execute(sql, (
            self.coord_start.lon, self.coord_start.lat,
            self.coord_end.lon,   self.coord_end.lat,
            RADAR_RANGE))

        while True:
            radar = st.fetchone()

            if radar is None:
                break

            radars.append(radar)

        return radars