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

Fix BUFRSounding

Browse source
This commit is contained in:
XANTRONIX 2025-04-12 11:25:50 -04:00
parent d55c83f54c
commit 86444caee6
1 changed files with 28 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

34
lib/xmet/bufr.py
View file

@ -1,9 +1,11 @@
from typing import Self
from awips.dataaccess import DataAccessLayer
from xmet.sounding import Sounding, SoundingSample
from xmet.units import celsius
class BUFRSounding():
class BUFRSounding(Sounding):
EDEX_HOST = 'edex-cloud.unidata.ucar.edu'
BUFR_TYPE = 'bufrua'
@ -11,12 +13,25 @@ class BUFRSounding():
BUFR_PARAMS_SIGT = set(['prSigT', 'htSigT', 'tpSigT', 'tdSigT'])
BUFR_PARAMS_SIGW = set(['prSigW', 'htSigW', 'wdSigW', 'wsSigW'])
def __init__(self):
super().__init__()
self.samples_by_pressure = dict()
def sample(self, pressure: float) -> SoundingSample:
sample = self.samples_by_pressure.get(pressure)
if sample is None:
self.samples_by_pressure[pressure] = SoundingSample()
return self.samples_by_pressure[pressure]
@staticmethod
def init():
DataAccessLayer.changeEDEXHost(BUFRSounding.EDEX_HOST)
@staticmethod
def latest(station: str) -> Sounding:
def latest(station: str) -> Self:
request = DataAccessLayer.newDataRequest()
request.setDatatype(BUFRSounding.BUFR_TYPE)
request.setParameters('staElev', 'staName')
@ -32,24 +47,31 @@ class BUFRSounding():
response = DataAccessLayer.getGeometryData(request,
times=datatimes[-1].validPeriod)
sounding = Sounding()
sounding = BUFRSounding()
for item in response:
params = item.getParameters()
if set(params) & BUFRSounding.BUFR_PARAMS_MAN:
sample = sounding.sample(item.getNumber('prMan') / 100.0)
pressure = item.getNumber('prMan') / 100.0
sample = sounding.sample(pressure)
sample.pressure = pressure
sample.height = item.getNumber('htMan')
sample.wind_speed = item.getNumber('wsMan')
sample.wind_dir = item.getNumber('wdMan')
if set(params) & BUFRSounding.BUFR_PARAMS_SIGT:
sample = sounding.sample(item.getNumber('prSigT') / 100.0)
pressure = item.getNumber('prSigT') / 100.0
sample = sounding.sample(pressure)
sample.temp = celsius(item.getNumber('tpSigT'))
sample.dewpoint = celsius(item.getNumber('tdSigT'))
if set(params) & BUFRSounding.BUFR_PARAMS_SIGW:
sample = sounding.sample(item.getNumber('prSigW') / 100.0)
pressure = item.getNumber('prSigW') / 100.0
sample = sounding.sample(pressure)
sample.height = item.getNumber('htSigW')
sample.wind_speed = item.getNumber('wsSigW')
sample.wind_dir = item.getNumber('wdSigW')