Implement RAOBSounding class

Implement RAOBSounding class to allow incremental parsing of a sounding
from separate observation blocks
This commit is contained in:
XANTRONIX 2025-03-27 16:46:58 -04:00
parent 2918a34621
commit 3b33faa78b

View file

@ -15,6 +15,85 @@ def meters_second(knots: float) -> float:
class RAOBReaderException(Exception): class RAOBReaderException(Exception):
... ...
class RAOBSounding():
DATA_SOURCE = 'UCAR'
station: str
timestamp: str
samples: dict[float, SoundingSample]
def __init__(self, station: str, timestamp: str):
self.station = station
self.timestamp = timestamp
self.samples = dict()
def parse_timestamp(self, token: str):
if token[0:2] == '//':
return None
day = int(token[0:2])
hour = int(token[2:4])
now = datetime.datetime.now(datetime.UTC)
return datetime.datetime(
year = now.year,
month = now.month,
day = day if day < 51 else day - 50,
hour = hour
)
def finish(self) -> Sounding:
timestamp = self.parse_timestamp(self.timestamp)
sounding = Sounding()
sounding.samples = list()
sounding.station = self.station
sounding.data_source_pressure = self.DATA_SOURCE
sounding.data_source_other = self.DATA_SOURCE
sounding.timestamp_observed = timestamp
sounding.timestamp_released = timestamp - datetime.timedelta(minutes=45)
for pressure in sorted(self.samples.keys(), reverse=True):
sounding.samples.append(self.samples[pressure])
return sounding
def sample(self, pressure: float):
if pressure in self.samples:
return self.samples[pressure]
sample = SoundingSample()
sample.pressure = pressure
sample.pressure_qa = ' '
sample.height_qa = ' '
sample.temp_qa = ' '
self.samples[pressure] = sample
return sample
def record_height(self, pressure: float, height: float):
sample = self.sample(pressure)
sample.height = height
def record_temp_dewpoint(self,
pressure: float,
temp: float,
dewpoint: float):
sample = self.sample(pressure)
sample.temp = temp
sample.dewpoint = dewpoint
def record_wind_speed_dir(self,
pressure: float,
wind_speed: float,
wind_dir: float):
sample = self.sample(pressure)
sample.wind_speed = wind_speed
sample.wind_dir = wind_dir
class RAOBObs(): class RAOBObs():
DATA_SOURCE = 'UCAR' DATA_SOURCE = 'UCAR'