Skip to content

Instantly share code, notes, and snippets.

@chomy
Last active October 10, 2020 09:37
Show Gist options
  • Save chomy/2d9a0bdd0b9d1e31e412ccb47d54db72 to your computer and use it in GitHub Desktop.
Save chomy/2d9a0bdd0b9d1e31e412ccb47d54db72 to your computer and use it in GitHub Desktop.
JMV3.0 parser
#!/usr/bin/python3
from datetime import datetime
from datetime import timedelta
import json
import sys
class JVMParser:
def parse_header(self, line, result):
row = tuple(filter(lambda a: a != '', line[0].split(' ')))
result['issued time'] = datetime.strptime(row[5] + 'GMT', '%y%m%d%H%M%S%Z').isoformat()
row = tuple(filter(lambda a:a!='', line[1].split(' ')))
result['title'] = 'Tropical Stom %s (%s) Warning #%s'%(row[1], row[2], 22)
result['number'] = row[1]
result['name'] = row[2]
result['time'] = datetime.strptime(row[0] + 'GMT', '%Y%m%d%H%M%Z')
return (line[2:], result)
def parse_forcast(self, lines, result):
def windarea_iterator(row, result):
if (len(row) == 0):
return result
area = {}
area['windspeed'] = int(row[0][1:])
area['radius'] = {}
area['radius'][row[2]] = int(row[1])
area['radius'][row[5]] = int(row[4])
area['radius'][row[8]] = int(row[7])
area['radius'][row[11]] = int(row[10])
result.append(area)
return windarea_iterator(row[13:], result)
def iterator(lines):
row = tuple(filter(lambda a: a != '', lines[0].split(' ')))
if (row[0].find('AMP') == 0):
return (lines, result)
r = {}
r['time'] = (result['time'] + timedelta(hours=int(row[0][1:]))).isoformat()
r['position'] = (row[1], row[2])
r['windspeed'] = int(row[3])
if (len(row) > 4):
r['windarea'] = []
r['windarea'].append(windarea_iterator(row[4:], []))
result['forcasts'].append(r)
return iterator(lines[1:])
result['forcasts'] = []
return iterator(lines)
def skip(self, lines, result):
for i in range(len(lines)):
row = tuple(filter(lambda a: a != '', lines[i].split(' ')))
if (len(row[0]) == 10 and row[0].isnumeric()):
return (lines[i:], result)
def parse_remarks(self, lines, result):
parse_pos = lambda a: '%.1f%s'%(int(a[:-1])/10, a[-1])
def iterator(lines, result):
row = tuple(filter(lambda a: a != '', lines[0].split(' ')))
if (row[0].find('NNNN') >= 0):
return (lines[1:], result)
r = {}
r['time'] = datetime.strptime(row[0][2:], '%y%m%d%H').isoformat()
r['position'] = (parse_pos(row[1][0:4]), parse_pos(row[1][5:]))
r['windspeed'] = int(row[2])
result['route'].append(r)
return iterator(lines[1:], result)
result['route'] = []
return iterator(lines[1:],result)
def parse(self, filename):
lines = []
with open(filename) as f:
for line in f:
lines.append(line[:-1])
self.result = {}
lines, result = self.parse_header(tuple(lines[1:]), {})
lines, result = self.parse_forcast(lines, result)
lines, result = self.skip(lines, result)
lines, result = self.parse_remarks(lines, result)
del result['time']
return result
def usage(argv):
print('Usage: %s filename'%argv[0])
def main(argv):
if (len(argv) != 2):
usage(argv)
sys.exit(-1)
p = JVMParser()
print(json.dumps(p.parse(argv[1])))
if (__name__ == '__main__'):
main(sys.argv)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment