143 lines
3.3 KiB
Python
143 lines
3.3 KiB
Python
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():
|
|
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):
|
|
start = parse_timestamp(text_start)
|
|
end = parse_timestamp(text_end)
|
|
|
|
if is_timestamp_zero(text_start) and not is_timestamp_zero(text_end):
|
|
self.timestamp_start = end
|
|
self.timestamp_end = end
|
|
elif not is_timestamp_zero(text_start) and is_timestamp_zero(text_end):
|
|
self.timestamp_start = start
|
|
self.timestamp_end = start
|
|
else:
|
|
self.timestamp_start = start
|
|
self.timestamp_end = end
|
|
|
|
class VTECEvent(VTECBaseEvent):
|
|
typeof: str
|
|
actions: str
|
|
wfo: str
|
|
phenom: str
|
|
sig: str
|
|
etn: int
|
|
|
|
timestamp_start: datetime.datetime
|
|
timestamp_end: datetime.datetime
|
|
|
|
@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', 'timestamp_start', 'timestamp_end'
|
|
)
|
|
|
|
severity: str
|
|
cause: str
|
|
record: str
|
|
|
|
timestamp_start: datetime.datetime
|
|
timestamp_end: datetime.datetime
|
|
|
|
@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
|