nexrad-archive/lib/nexrad/vtec.py

import re
import enum
import datetime

from typing import Self

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)

def is_timestamp_zero(text: str):
    return text == '000000T0000Z'

def parse_timestamp(text: str):
    year   = int(text[0:2])
    month  = int(text[2:4])
    day    = int(text[4:6])
    hour   = int(text[7:9])
    minute = int(text[9:11])

    if year >= 70:
        year += 1900
    else:
        year += 2000

    if month == 0:
        month = 1

    if day == 0:
        day = 1

    return datetime.datetime(
        year   = year,
        month  = month,
        day    = day,
        hour   = hour,
        minute = minute,
        second = 0,
        tzinfo = datetime.UTC
    )

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

class VTECBaseEvent():
    __slots__ = 'timestamp_start', 'timestamp_end',

    timestamp_start: datetime.datetime
    timestamp_end:   datetime.datetime

    def __init__(self):
        self.timestamp_start = None
        self.timestamp_end   = None

    def parse_timestamps(self, text_start: str, text_end: str):
        if not is_timestamp_zero(text_start):
            self.timestamp_start = parse_timestamp(text_start)

        if not is_timestamp_zero(text_end):
            self.timestamp_end = parse_timestamp(text_end)

class VTECEvent(VTECBaseEvent):
    __slots__ = (
        'typeof', 'actions', 'wfo', 'phenom', 'sig', 'etn'
    )

    typeof:  str
    actions: str
    wfo:     str
    phenom:  str
    sig:     str
    etn:     int

    @staticmethod
    def parse(text: str) -> Self:
        match = RE_PHENOM.match(text)

        if match is None:
            return

        event = VTECEvent()
        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'])

        event.parse_timestamps(match['time_start'], match['time_end'])

        return event

class VTECHydroEvent(VTECBaseEvent):
    __slots__ = (
        'severity', 'cause', 'record'
    )

    severity: str
    cause:    str
    record:   str

    @staticmethod
    def parse(text: str) -> Self:
        match = RE_HYDRO.match(text)

        if match is None:
            return

        event = VTECHydroEvent()
        event.severity = match['severity']
        event.cause    = match['cause']
        event.record   = match['record']

        event.parse_timestamps(match['time_start'], match['time_end'])

        return event