Refactor parser into AFOSMessageParser
Refactor parser into AFOSMessageParser to reduce cyclomatic complexity
This commit is contained in:
XANTRONIX Industrial
parent
451a63d067
commit
52978b267f
1 changed files with 137 additions and 108 deletions
|
|
@ -167,116 +167,145 @@ class AFOSMessage(DatabaseTable):
|
||||||
self.forecaster = None
|
self.forecaster = None
|
||||||
self.poly = None
|
self.poly = None
|
||||||
|
|
||||||
@staticmethod
|
def is_watch(self):
|
||||||
def parse(text: str) -> Self:
|
return self.sig is not None and self.sig == 'A'
|
||||||
message = AFOSMessage()
|
|
||||||
message.text_raw = text
|
|
||||||
|
|
||||||
state = AFOSMessageParserState.SERIAL
|
def is_warning(self):
|
||||||
|
return self.sig is not None and self.sig == 'W'
|
||||||
|
|
||||||
issuance = None
|
class AFOSMessageParser():
|
||||||
timestamp_inline = None
|
__slots__ = 'message', 'state', 'issuance', 'timestamp',
|
||||||
|
|
||||||
for line in text.split('\n'):
|
def __init__(self):
|
||||||
line = line.rstrip()
|
self.message = None
|
||||||
|
self.state = None
|
||||||
|
self.issuance = None
|
||||||
|
self.timestamp = None
|
||||||
|
|
||||||
if line == '':
|
def parse_vtec(self, line: str):
|
||||||
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)
|
vtec = VTECEvent.parse(line)
|
||||||
|
|
||||||
if vtec is not None:
|
if vtec is not None:
|
||||||
message.timestamp_start = vtec.timestamp_start
|
self.message.timestamp_start = vtec.timestamp_start
|
||||||
message.timestamp_end = vtec.timestamp_end
|
self.message.timestamp_end = vtec.timestamp_end
|
||||||
|
|
||||||
message.vtec_type = vtec.typeof
|
self.message.vtec_type = vtec.typeof
|
||||||
message.actions = vtec.actions
|
self.message.actions = vtec.actions
|
||||||
message.wfo = vtec.wfo
|
self.message.wfo = vtec.wfo
|
||||||
message.phenom = vtec.phenom
|
self.message.phenom = vtec.phenom
|
||||||
message.sig = vtec.sig
|
self.message.sig = vtec.sig
|
||||||
message.etn = vtec.etn
|
self.message.etn = vtec.etn
|
||||||
|
|
||||||
if state == AFOSMessageParserState.SERIAL:
|
def parse_serial(self, line: str):
|
||||||
match = RE_ID.match(line)
|
match = RE_ID.match(line)
|
||||||
|
|
||||||
if match is not None:
|
if match is not None:
|
||||||
message.serial = int(match[1])
|
self.message.serial = int(match[1])
|
||||||
state = AFOSMessageParserState.ISSUANCE
|
self.state = AFOSMessageParserState.ISSUANCE
|
||||||
elif state == AFOSMessageParserState.ISSUANCE:
|
|
||||||
|
def parse_issuance(self, line: str):
|
||||||
match = RE_ISSUANCE.match(line)
|
match = RE_ISSUANCE.match(line)
|
||||||
|
|
||||||
if match is not None:
|
if match is not None:
|
||||||
state = AFOSMessageParserState.PRODUCT
|
self.issuance = match
|
||||||
issuance = match
|
self.state = AFOSMessageParserState.PRODUCT
|
||||||
elif state == AFOSMessageParserState.PRODUCT:
|
|
||||||
|
def parse_product(self, line: str):
|
||||||
match = RE_PRODUCT.match(line)
|
match = RE_PRODUCT.match(line)
|
||||||
|
|
||||||
if match is not None:
|
if match is not None:
|
||||||
message.product = match['product']
|
self.message.product = match['product']
|
||||||
|
self.state = AFOSMessageParserState.BODY
|
||||||
|
|
||||||
state = AFOSMessageParserState.BODY
|
def parse_body(self, line: str):
|
||||||
elif state == AFOSMessageParserState.BODY:
|
if line == '&&':
|
||||||
if timestamp_inline is None:
|
self.state = AFOSMessageParserState.TAGS
|
||||||
|
elif self.timestamp is None:
|
||||||
match = RE_DATE.match(line)
|
match = RE_DATE.match(line)
|
||||||
|
|
||||||
if match is not None:
|
if match is not None:
|
||||||
offset = TIMEZONES[match['tz'].upper()]
|
tzoffset = TIMEZONES[match['tz'].upper()]
|
||||||
timestamp_inline = datetime.datetime(
|
|
||||||
|
self.timestamp = datetime.datetime(
|
||||||
year = int(match['year']),
|
year = int(match['year']),
|
||||||
month = MONTHS[match['month'].upper()],
|
month = MONTHS[match['month'].upper()],
|
||||||
day = int(match['day']),
|
day = int(match['day']),
|
||||||
hour = int(match['hour']),
|
hour = int(match['hour']),
|
||||||
minute = int(match['minute']),
|
minute = int(match['minute']),
|
||||||
second = 0,
|
second = 0,
|
||||||
tzinfo = datetime.timezone(datetime.timedelta(hours=offset))
|
tzinfo = datetime.timezone(datetime.timedelta(hours=tzoffset))
|
||||||
).astimezone(datetime.UTC)
|
).astimezone(datetime.UTC)
|
||||||
|
|
||||||
if line == '&&':
|
def parse_tags(self, line: str):
|
||||||
state = AFOSMessageParserState.TAGS
|
|
||||||
elif state == AFOSMessageParserState.TAGS:
|
|
||||||
if line == '$$':
|
if line == '$$':
|
||||||
state = AFOSMessageParserState.FOOTER
|
self.state = AFOSMessageParserState.FOOTER
|
||||||
else:
|
return
|
||||||
|
|
||||||
match = RE_POLY.match(line)
|
match = RE_POLY.match(line)
|
||||||
|
|
||||||
if match is not None:
|
if match is not None:
|
||||||
message.poly = parse_shape(match['coords'])
|
self.message.poly = parse_shape(match['coords'])
|
||||||
|
|
||||||
match = RE_MOTION.match(line)
|
match = RE_MOTION.match(line)
|
||||||
|
|
||||||
if match is not None:
|
if match is not None:
|
||||||
message.azimuth = int(match['azimuth'])
|
self.message.azimuth = int(match['azimuth'])
|
||||||
message.speed = int(match['speed'])
|
self.message.speed = int(match['speed'])
|
||||||
message.location = parse_location(match['lon'], match['lat'])
|
self.message.location = parse_location(match['lon'], match['lat'])
|
||||||
elif state == AFOSMessageParserState.FOOTER:
|
|
||||||
if line != '':
|
|
||||||
message.forecaster = line
|
|
||||||
|
|
||||||
if message.timestamp_issued is None:
|
def parse_footer(self, line: str):
|
||||||
if timestamp_inline is not None:
|
self.message.forecaster = line
|
||||||
message.timestamp_issued = timestamp_inline
|
|
||||||
message.timestamp_start = timestamp_inline
|
def parse_line(self, line: str):
|
||||||
message.timestamp_end = timestamp_inline + datetime.timedelta(hours=1)
|
if line == '':
|
||||||
else:
|
return
|
||||||
message.timestamp_issued = datetime.datetime(
|
elif line[0] == '/' and line[-1] == '/':
|
||||||
year = message.timestamp_start.year,
|
#
|
||||||
month = message.timestamp_start.month,
|
# The VTEC line can appear anywhere in the message
|
||||||
day = int(issuance['day']),
|
# text, therefore, parsing must be able to occur in
|
||||||
hour = int(issuance['hour']),
|
# all states.
|
||||||
minute = int(issuance['minute']),
|
#
|
||||||
|
self.parse_vtec(line)
|
||||||
|
|
||||||
|
if self.state == AFOSMessageParserState.SERIAL:
|
||||||
|
self.parse_serial(line)
|
||||||
|
elif self.state == AFOSMessageParserState.ISSUANCE:
|
||||||
|
self.parse_issuance(line)
|
||||||
|
elif self.state == AFOSMessageParserState.PRODUCT:
|
||||||
|
self.parse_product(line)
|
||||||
|
elif self.state == AFOSMessageParserState.BODY:
|
||||||
|
self.parse_body(line)
|
||||||
|
elif self.state == AFOSMessageParserState.TAGS:
|
||||||
|
self.parse_tags(line)
|
||||||
|
elif self.state == AFOSMessageParserState.FOOTER:
|
||||||
|
self.parse_footer(line)
|
||||||
|
|
||||||
|
def assign_timestamps(self):
|
||||||
|
if self.message.timestamp_issued is None:
|
||||||
|
if self.timestamp is not None:
|
||||||
|
self.message.timestamp_issued = self.timestamp
|
||||||
|
self.message.timestamp_start = self.timestamp
|
||||||
|
self.message.timestamp_end = self.timestamp + datetime.timedelta(hours=1)
|
||||||
|
elif self.issuance is not None:
|
||||||
|
self.message.timestamp_issued = datetime.datetime(
|
||||||
|
year = self.message.timestamp_start.year,
|
||||||
|
month = self.message.timestamp_start.month,
|
||||||
|
day = int(self.issuance['day']),
|
||||||
|
hour = int(self.issuance['hour']),
|
||||||
|
minute = int(self.issuance['minute']),
|
||||||
second = 0,
|
second = 0,
|
||||||
tzinfo = datetime.UTC
|
tzinfo = datetime.UTC
|
||||||
)
|
)
|
||||||
|
|
||||||
return message
|
def parse(self, text: str):
|
||||||
|
self.message = AFOSMessage()
|
||||||
|
self.state = AFOSMessageParserState.SERIAL
|
||||||
|
self.issuance = None
|
||||||
|
self.timestamp = None
|
||||||
|
|
||||||
def is_watch(self):
|
for line in text.split('\n'):
|
||||||
return self.sig is not None and self.sig == 'A'
|
self.parse_line(line.rstrip())
|
||||||
|
|
||||||
def is_warning(self):
|
self.assign_timestamps()
|
||||||
return self.sig is not None and self.sig == 'W'
|
|
||||||
|
return self.message
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue