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

Make headway in parsing RAOB TTBB reports

Browse source
This commit is contained in:
XANTRONIX 2025-03-28 16:14:44 -04:00
parent 32e71c888c
commit b7bee03e85
1 changed files with 47 additions and 46 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

93
lib/xmet/raob.py
View file

@ -258,38 +258,27 @@ class RAOBObs():
'height': height 'height': height
} }
PRESSURE_SIG = { PRESSURE_CODES = {
'11': True, '22': True, '33': True, '44': True, '55': True, '11': True, '22': True, '33': True, '44': True, '55': True,
'66': True, '77': True, '88': True, '99': True '66': True, '77': True, '88': True, '99': True
} }
def parse_significant_pressure(self, token: str): def parse_pressure(self, token: str):
code, pressure = token[0:2], token[2:5]
if code in self.PRESSURE_SIG:
return {
'height': None,
'pressure': float(pressure)
}
def parse_surface_pressure(self, token: str):
code, pressure = token[0:2], float(token[2:5]) code, pressure = token[0:2], float(token[2:5])
if pressure < 100: if pressure < 100:
pressure += 1000.0 pressure += 1000.0
if code == '99': return {
return { 'pressure': pressure if code in self.PRESSURE_CODES else None,
'height': None, 'height': None
'pressure': pressure }
}
def parse_sample_tokens(self, tokens: list[str]) -> SoundingSample: def parse_ttaa_sample(self, tokens: list[str]) -> dict:
sample = SoundingSample() sample = SoundingSample()
if tokens[0][0:2] == '99': if tokens[0][0:2] == '99':
sample.elapsed = 0 hp = self.parse_pressure(tokens[0])
hp = self.parse_surface_pressure(tokens[0])
else: else:
hp = self.parse_height_pressure(tokens[0]) hp = self.parse_height_pressure(tokens[0])
@ -299,16 +288,14 @@ class RAOBObs():
td = self.parse_temp_dewpoint(tokens[1]) td = self.parse_temp_dewpoint(tokens[1])
wind = self.parse_wind(tokens[2]) wind = self.parse_wind(tokens[2])
sample.height = hp['height'] return {
sample.pressure = hp['pressure'] 'pressure': hp['pressure'],
sample.temp = td['temp'] if td is not None else None 'height': hp['height'],
sample.dewpoint = td['dewpoint'] if td is not None else None 'temp': td['temp'],
sample.wind_dir = wind['dir'] if wind is not None else None 'dewpoint': td['dewpoint'],
sample.wind_speed = wind['speed'] if wind is not None else None 'wind_speed': wind['speed'],
'wind_dir': wind['dir']
sample.pressure_qa = ' ' }
sample.height_qa = ' '
sample.temp_qa = ' '
return sample return sample
@ -325,16 +312,11 @@ class RAOBObs():
if self.tokens[1][0:3] == 'NIL': if self.tokens[1][0:3] == 'NIL':
return None return None
sample = self.parse_sample_tokens(self.tokens[2:5])
if sample is None:
return
station = self.tokens[1] station = self.tokens[1]
timestamp = self.tokens[0] timestamp = self.tokens[0]
samples = list() samples = list()
for i in range(5, len(self.tokens), 3): for i in range(2, len(self.tokens), 3):
if len(self.tokens) < i+3 or self.tokens[i][-1] == '=': if len(self.tokens) < i+3 or self.tokens[i][-1] == '=':
break break
@ -344,12 +326,31 @@ class RAOBObs():
if self.tokens[i][0:2] == '88': if self.tokens[i][0:2] == '88':
break break
sample = self.parse_sample_tokens(self.tokens[i:i+3]) samples.append(self.parse_ttaa_sample(self.tokens[i:i+3]))
if sample is None: return {
continue 'station': station,
'timestamp': timestamp,
'samples': samples
}
samples.append(sample) def parse_ttbb_sample(self, tokens: list[str]) -> dict:
hp = self.parse_pressure(tokens[0])
td = self.parse_temp_dewpoint(tokens[1])
return {
'pressure': hp['pressure'],
'temp': td['temp'],
'dewpoint': td['dewpoint']
}
def parse_ttbb(self) -> dict:
station = self.tokens[1]
timestamp = self.tokens[2]
samples = list()
for i in range(3, len(self.tokens), 2):
samples.append(self.parse_sample_tokens(self.tokens[i:i+2]))
return { return {
'station': station, 'station': station,
@ -499,18 +500,18 @@ class RAOBReader():
sounding = self.sounding(station, timestamp) sounding = self.sounding(station, timestamp)
for sample in samples: for data in samples:
pressure = sample.pressure pressure = data['pressure']
sounding.record_height(pressure, sample.height) sounding.record_height(pressure, data['height'])
sounding.record_temp_dewpoint(pressure, sounding.record_temp_dewpoint(pressure,
sample.temp, data['temp'],
sample.dewpoint) data['dewpoint'])
sounding.record_wind_speed_dir(pressure, sounding.record_wind_speed_dir(pressure,
sample.wind_speed, data['wind_speed'],
sample.wind_dir) data['wind_dir'])
for key in self.soundings: for key in self.soundings:
yield self.soundings[key].finish() yield self.soundings[key].finish()