282 lines
8.6 KiB
Python
282 lines
8.6 KiB
Python
import re
|
|
import enum
|
|
import datetime
|
|
import shapely
|
|
|
|
from typing import Self
|
|
|
|
from nexrad.db import DatabaseTable
|
|
from nexrad.coord import COORD_SYSTEM
|
|
from nexrad.vtec import VTECEvent
|
|
|
|
RE_ID = re.compile(r'^(\d+)$')
|
|
|
|
RE_ISSUANCE = re.compile(r'''
|
|
^ ([A-Z]{4}\d+)
|
|
\s+ (?P<wfo>[A-Z]{4})
|
|
\s+ (?P<day>\d{2}) (?P<hour>\d{2}) (?P<minute>\d{2})
|
|
''', re.X)
|
|
|
|
RE_DATE = re.compile(r'''
|
|
^ (?P<hour>\d{1,2})
|
|
(?P<minute>\d{2})
|
|
\s+ (AM|PM)
|
|
\s+ (?P<tz>[A-Z]{3})
|
|
\s+ (?P<weekday>[A-Za-z]+)
|
|
\s+ (?P<month>[A-Za-z]+)
|
|
\s+ (?P<day>\d{1,2})
|
|
\s+ (?P<year>\d{4})
|
|
''', re.X)
|
|
|
|
RE_PRODUCT = re.compile(r'^(?P<product>[A-Z]{3})(?P<wfo>[A-Z]{3})$')
|
|
|
|
RE_POLY = re.compile(r'^LAT\.\.\.LON (?P<coords>\d+(?: \d+)+)')
|
|
|
|
RE_MOTION = re.compile(r'''
|
|
^ TIME
|
|
\.\.\. MOT
|
|
\.\.\. LOC
|
|
\s+ (?P<hour>\d{2})(?P<minute>\d{2})Z
|
|
\s+ (?P<azimuth>\d+)DEG
|
|
\s+ (?P<speed>\d+)KT
|
|
\s+ (?P<lat>\d+)
|
|
\s+ (?P<lon>\d+)
|
|
$
|
|
''', re.X)
|
|
|
|
MONTHS = {
|
|
'JAN': 1, 'FEB': 2, 'MAR': 3, 'APR': 4, 'MAY': 5, 'JUN': 6,
|
|
'JUL': 7, 'AUG': 8, 'SEP': 9, 'OCT': 10, 'NOV': 11, 'DEC': 12,
|
|
|
|
'JANUARY': 1, 'FEBRUARY': 2, 'MARCH': 3, 'APRIL': 4,
|
|
'MAY': 5, 'JUNE': 6, 'JULY': 7, 'AUGUST': 8,
|
|
'SEPTEMBER': 9, 'OCTOBER': 10, 'NOVEMBER': 11, 'DECEMBER': 12
|
|
}
|
|
|
|
TIMEZONES = {
|
|
'HST': -10, 'PST': -8, 'PDT': -7, 'MST': -7, 'MDT': -6, 'CST': -6,
|
|
'CDT': -5, ' EST': -5, 'EDT': -4, 'GMT': 0
|
|
}
|
|
|
|
def parse_lon(text: str):
|
|
size = len(text)
|
|
return 0 - float(text[0:size-2]) + (float(text[size-2:size]) / 100)
|
|
|
|
def parse_lat(text: str):
|
|
size = len(text)
|
|
return float(text[0:size-2]) + (float(text[size-2:size]) / 100)
|
|
|
|
def parse_location(lon: str, lat: str):
|
|
return shapely.Point(parse_lon(lon), parse_lat(lat))
|
|
|
|
def parse_shape(text: str):
|
|
points = list()
|
|
coords = text.split(' ')
|
|
|
|
for i in range(0, len(coords), 2):
|
|
lat = coords[i]
|
|
lon = coords[i+1]
|
|
|
|
points.append([parse_lon(lon), parse_lat(lat)])
|
|
|
|
points.append([parse_lon(coords[0]), parse_lat(coords[1])])
|
|
|
|
return shapely.Polygon(points)
|
|
|
|
class AFOSMessageParserState(enum.Enum):
|
|
NONE = 0
|
|
SERIAL = enum.auto()
|
|
ISSUANCE = enum.auto()
|
|
PRODUCT = enum.auto()
|
|
BODY = enum.auto()
|
|
TAGS = enum.auto()
|
|
FOOTER = enum.auto()
|
|
|
|
class AFOSMessage(DatabaseTable):
|
|
__table__ = 'nexrad_afos_message'
|
|
__key__ = 'id'
|
|
|
|
__columns__ = (
|
|
'id', 'timestamp_issued', 'timestamp_start', 'timestamp_end',
|
|
'serial', 'product', 'vtec_type', 'etn', 'actions', 'wfo',
|
|
'phenom', 'sig', 'text_raw', 'azimuth', 'speed', 'location',
|
|
'forecaster', 'poly',
|
|
)
|
|
|
|
__columns_read__ = {
|
|
'poly': 'ST_AsText(poly) as poly',
|
|
'location': 'ST_AsText(location) as location'
|
|
}
|
|
|
|
__values_write__ = {
|
|
'poly': shapely.from_wkt,
|
|
'location': shapely.from_wkt
|
|
}
|
|
|
|
__columns_write__ = {
|
|
'poly': 'ST_GeomFromText(:poly, {crs})'.format(crs=COORD_SYSTEM),
|
|
'location': 'ST_GeomFromText(:location, {crs})'.format(crs=COORD_SYSTEM)
|
|
}
|
|
|
|
__values_write__ = {
|
|
'poly': lambda v: {'poly': shapely.to_wkt(v)},
|
|
'location': lambda v: {'location': shapely.to_wkt(v)}
|
|
}
|
|
|
|
id: int
|
|
serial: int
|
|
|
|
timestamp_issued: datetime.datetime
|
|
timestamp_start: datetime.datetime
|
|
timestamp_end: datetime.datetime
|
|
|
|
product: str
|
|
vtec_type: str
|
|
actions: str
|
|
wfo: str
|
|
phenom: str
|
|
sig: str
|
|
etn: int
|
|
text_raw: str
|
|
azimuth: int
|
|
speed: int
|
|
location: shapely.Point
|
|
forecaster: str
|
|
poly: shapely.Geometry
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.id = None
|
|
self.serial = None
|
|
|
|
self.timestamp_issued = None
|
|
self.timestamp_start = None
|
|
self.timestamp_end = None
|
|
|
|
self.product = None
|
|
self.vtec_type = None
|
|
self.actions = None
|
|
self.wfo = None
|
|
self.phenom = None
|
|
self.sig = None
|
|
self.etn = None
|
|
self.text_raw = None
|
|
self.azimuth = None
|
|
self.speed = None
|
|
self.location = None
|
|
self.forecaster = None
|
|
self.poly = None
|
|
|
|
@staticmethod
|
|
def parse(text: str) -> Self:
|
|
message = AFOSMessage()
|
|
message.text_raw = text
|
|
|
|
state = AFOSMessageParserState.SERIAL
|
|
|
|
issuance = None
|
|
timestamp_inline = None
|
|
|
|
for line in text.split('\n'):
|
|
line = line.rstrip()
|
|
|
|
if line == '':
|
|
continue
|
|
elif line[0] == '/' and line[-1] == '/':
|
|
#
|
|
# The VTEC line can appear anywhere in the message
|
|
# text, therefore, parsing must be able to occur in
|
|
# all states.
|
|
#
|
|
vtec = VTECEvent.parse(line)
|
|
|
|
if vtec is not None:
|
|
message.timestamp_start = vtec.timestamp_start
|
|
message.timestamp_end = vtec.timestamp_end
|
|
|
|
message.vtec_type = vtec.typeof
|
|
message.actions = vtec.actions
|
|
message.wfo = vtec.wfo
|
|
message.phenom = vtec.phenom
|
|
message.sig = vtec.sig
|
|
message.etn = vtec.etn
|
|
|
|
if state == AFOSMessageParserState.SERIAL:
|
|
match = RE_ID.match(line)
|
|
|
|
if match is not None:
|
|
message.serial = int(match[1])
|
|
state = AFOSMessageParserState.ISSUANCE
|
|
elif state == AFOSMessageParserState.ISSUANCE:
|
|
match = RE_ISSUANCE.match(line)
|
|
|
|
if match is not None:
|
|
state = AFOSMessageParserState.PRODUCT
|
|
issuance = match
|
|
elif state == AFOSMessageParserState.PRODUCT:
|
|
match = RE_PRODUCT.match(line)
|
|
|
|
if match is not None:
|
|
message.product = match['product']
|
|
|
|
state = AFOSMessageParserState.BODY
|
|
elif state == AFOSMessageParserState.BODY:
|
|
if timestamp_inline is None:
|
|
match = RE_DATE.match(line)
|
|
|
|
if match is not None:
|
|
offset = TIMEZONES[match['tz'].upper()]
|
|
timestamp_inline = datetime.datetime(
|
|
year = int(match['year']),
|
|
month = MONTHS[match['month'].upper()],
|
|
day = int(match['day']),
|
|
hour = int(match['hour']),
|
|
minute = int(match['minute']),
|
|
second = 0,
|
|
tzinfo = datetime.timezone(datetime.timedelta(hours=offset))
|
|
).astimezone(datetime.UTC)
|
|
|
|
if line == '&&':
|
|
state = AFOSMessageParserState.TAGS
|
|
elif state == AFOSMessageParserState.TAGS:
|
|
if line == '$$':
|
|
state = AFOSMessageParserState.FOOTER
|
|
else:
|
|
match = RE_POLY.match(line)
|
|
|
|
if match is not None:
|
|
message.poly = parse_shape(match['coords'])
|
|
|
|
match = RE_MOTION.match(line)
|
|
|
|
if match is not None:
|
|
message.azimuth = int(match['azimuth'])
|
|
message.speed = int(match['speed'])
|
|
message.location = parse_location(match['lon'], match['lat'])
|
|
elif state == AFOSMessageParserState.FOOTER:
|
|
if line != '':
|
|
message.forecaster = line
|
|
|
|
if message.timestamp_issued is None:
|
|
if timestamp_inline is not None:
|
|
message.timestamp_issued = timestamp_inline
|
|
message.timestamp_start = timestamp_inline
|
|
message.timestamp_end = timestamp_inline + datetime.timedelta(hours=1)
|
|
else:
|
|
message.timestamp_issued = datetime.datetime(
|
|
year = message.timestamp_start.year,
|
|
month = message.timestamp_start.month,
|
|
day = int(issuance['day']),
|
|
hour = int(issuance['hour']),
|
|
minute = int(issuance['minute']),
|
|
second = 0,
|
|
tzinfo = datetime.UTC
|
|
)
|
|
|
|
return message
|
|
|
|
def is_watch(self):
|
|
return self.sig is not None and self.sig == 'A'
|
|
|
|
def is_warning(self):
|
|
return self.sig is not None and self.sig == 'W'
|