xan/xmet
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Implement parsing motion, time, azimuth tag

Browse source
This commit is contained in:
XANTRONIX Industrial 2025-02-19 16:55:37 -05:00
parent 928fe6c71e
commit 26a464e5e0
1 changed files with 43 additions and 11 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

54
lib/nexrad/vtec.py
View file

@ -54,7 +54,27 @@ RE_HYDRO = re.compile(r'''
/$
''', re.X)
RE_POLY = re.compile(r'^LAT\.\.\.LON (?P<coords>\d+(?: \d+)+)')
RE_POLY = re.compile(r'^LAT\.\.\.LON (?P<coords>\d+(?: \d+)+)')
RE_MOTION = re.compile(r'''
^
TIME
\.\.\.
MOT
\.\.\.
LOC
[ ]{1}
(?P<hour>\d{2})(?P<minute>\d{2})Z
[ ]{1}
(?P<azimuth>\d+)DEG
[ ]{1}
(?P<speed>\d+)KT
[ ]{1}
(?P<lat>\d+)
[ ]{1}
(?P<lon>\d+)
$
''', re.X)
def parse_timestamp(text: str, post_2016_05_11: bool):
return datetime.datetime.strptime(
@ -69,6 +89,9 @@ 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(' ')
@ -98,7 +121,7 @@ class VTECEventParserState(enum.Enum):
VTEC = enum.auto()
BODY_SEP = enum.auto()
BODY = enum.auto()
POLY = enum.auto()
TAGS = enum.auto()
FOOTER = enum.auto()
class VTECEvent(DatabaseTable):
@ -108,7 +131,7 @@ class VTECEvent(DatabaseTable):
__columns__ = (
'id', 'timestamp_issued', 'timestamp_start', 'timestamp_end',
'typeof', 'etn', 'actions', 'wfo', 'phenom', 'sig', 'body',
'forecaster', 'poly',
'azimuth', 'speed', 'location', 'forecaster', 'poly',
)
__columns_read__ = {
@ -138,6 +161,9 @@ class VTECEvent(DatabaseTable):
sig: str
etn: int
body: str
azimuth: int
speed: int
location: shapely.Point
forecaster: str
poly: shapely.Geometry
@ -196,18 +222,24 @@ class VTECEvent(DatabaseTable):
state = VTECEventParserState.BODY
elif state == VTECEventParserState.BODY:
if line == '&&':
state = VTECEventParserState.POLY
state = VTECEventParserState.TAGS
else:
event.body += '\n' + line
elif state == VTECEventParserState.POLY:
match = RE_POLY.match(line)
if match is not None:
event.poly = parse_shape(match['coords'])
elif line == '$$':
elif state == VTECEventParserState.TAGS:
if line == '$$':
state = VTECEventParserState.FOOTER
else:
pass
match = RE_POLY.match(line)
if match is not None:
event.poly = parse_shape(match['coords'])
match = RE_MOTION.match(line)
if match is not None:
event.azimuth = int(match['azimuth'])
event.speed = int(match['speed'])
event.location = parse_location(match['lon'], match['lat'])
elif state == VTECEventParserState.FOOTER:
if line != '':
event.forecaster = line