Last active
October 10, 2020 09:37
-
-
Save chomy/2d9a0bdd0b9d1e31e412ccb47d54db72 to your computer and use it in GitHub Desktop.
JMV3.0 parser
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/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