nexrad-archive/lib/nexrad/vtec.py

import re
import enum
import datetime
import json
import shapely

from nexrad.db    import DatabaseTable
from nexrad.coord import COORD_SYSTEM

RE_ID = re.compile(r'^(\d+)$')

RE_ISSUANCE = re.compile(r'''
    ^
    (WF[A-Z]{2}\d{2})
    [ ]{1}
    (?P<wfo>[A-Z]{4})
    [ ]{1}
    (?P<day>\d{2})
    (?P<hour>\d{2})
    (?P<minute>\d{2})
    $
''', re.X)

RE_PHENOM = re.compile(r'''
    ^/
    (?P<typeof>[OTEX])
    \.
    (?P<actions>[A-Z]{3})
    \.
    (?P<wfo>[A-Z]{4})
    \.
    (?P<phenom>[A-Z]{2})
    \.
    (?P<sig>[A-Z])
    \.
    (?P<etn>\d{4})
    \.
    (?P<time_start>\d{6}T\d{4}Z)
    -
    (?P<time_end>\d{6}T\d{4}Z)
    /$
''', re.X)

RE_HYDRO = re.compile(r'''
    ^/
    (?P<severity>[0N1])
    \.
    (?P<cause>[A-Z]{2})
    \.
    (?P<time_start>\d{6}T\d{4}Z)
    -
    (?P<time_end>\d{6}T\d{4}Z)
    \.
    (?P<record>[A-Z]{2})
    /$
''', re.X)

RE_POLY = re.compile(r'^LAT\.\.\.LON(?P<coords> \d{4})+')

def parse_timestamp(text: str, post_2016_05_11: bool):
    return datetime.datetime.strptime(
        text, '%y%m%dT%H%M%SZ'
    ).astimezone(datetime.UTC)

class VTECEventType(enum.StrEnum):
    OPERATIONAL       = 'O'
    TEST              = 'T'
    EXPERIMENTAL      = 'E'
    EXPERIMENTAL_VTEC = 'X'

class VTECEventParserState(enum.Enum):
    NONE       = 1
    HEADER     = enum.auto()
    ISSUANCE   = enum.auto()
    META       = enum.auto()
    TYPEOFFICE = enum.auto()
    VTEC       = enum.auto()
    BODY_SEP   = enum.auto()
    BODY       = enum.auto()
    POLY       = enum.auto()
    FOOTER     = enum.auto()

class VTECEvent(DatabaseTable):
    __table__ = 'nexrad_vtec_event'
    __key__   = 'id'

    __columns__ = (
        'id', 'timestamp_issued', 'timestamp_start', 'timestamp_end',
        'typeof', 'etn', 'actions', 'wfo', 'phenom', 'sig', 'body',
        'forecaster', 'poly',
    )

    __columns_read__ = {
        'poly': 'ST_AsText(poly) as poly'
    }

    __values_write__ = {
        'poly': shapely.from_wkt
    }

    __columns_write__ = {
        'poly': 'ST_GeomFromText(:poly, {crs})'.format(crs=COORD_SYSTEM)
    }

    __values_write__ = {
        'poly': lambda v: {'poly': shapely.to_wkt(v)}
    }

    id: int
    timestamp_issued: datetime.datetime
    timestamp_start:  datetime.datetime
    timestamp_end:    datetime.datetime
    typeof:           str
    actions:          str
    wfo:              str
    phenom:           str
    sig:              str
    etn:              int
    body:             str
    forecaster:       str
    poly:             shapely.Geometry

    def parse_shape(self, coords: str):
        pass

    @staticmethod
    def parse(text: str):
        event = VTECEvent()
        state = VTECEventParserState.NONE

        #
        # A timestamp post 11 May 2016 can be detected based on the
        # presence of lowercase letters in bulletin text, as per:
        #
        # https://www.noaa.gov/media-release/national-weather-service-will-stop-using-all-caps-in-its-forecasts
        #
        post_2016_05_11 = any(c for c in text if c.islower())

        issuance = None

        for line in text.split('\n'):
            line = line.rstrip()

            if state == VTECEventParserState.NONE:
                match = RE_ID.match(line)

                if match is not None:
                    event.id = int(match[1])
                    state = VTECEventParserState.HEADER
            elif state == VTECEventParserState.HEADER:
                match = RE_ISSUANCE.match(line)

                if match is not None:
                    issuance = match
                    state = VTECEventParserState.ISSUANCE
            elif state == VTECEventParserState.ISSUANCE:
                state = VTECEventParserState.META
            elif state == VTECEventParserState.META:
                match = RE_PHENOM.match(line)

                if match is not None:
                    event.timestamp_start = parse_timestamp(match['time_start'], post_2016_05_11)
                    event.timestamp_end   = parse_timestamp(match['time_end'], post_2016_05_11)

                    event.typeof  = match['typeof']
                    event.actions = match['actions']
                    event.wfo     = match['wfo']
                    event.phenom  = match['phenom']
                    event.sig     = match['sig']
                    event.etn     = int(match['etn'])

                    state = VTECEventParserState.VTEC
            elif state == VTECEventParserState.VTEC:
                if line == '':
                    state = VTECEventParserState.BODY_SEP
            elif state == VTECEventParserState.BODY_SEP:
                event.body = line
                state = VTECEventParserState.BODY
            elif state == VTECEventParserState.BODY:
                if line == '&&':
                    state = VTECEventParserState.POLY
                else:
                    event.body += '\n' + line
            elif state == VTECEventParserState.POLY:
                match = RE_POLY.match(line)

                if match is not None:
                    event.parse_shape(match['coords'])
                elif line == '$$':
                    state = VTECEventParserState.FOOTER
                else:
                    pass
            elif state == VTECEventParserState.FOOTER:
                if line != '':
                    event.forecaster = line

        return event