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

Split BUFRSounding.latest() into separate methods

Browse source
This commit is contained in:
XANTRONIX 2025-04-14 11:04:33 -04:00
parent 854cfa65e4
commit 17df1e1dba
1 changed files with 16 additions and 8 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

24
lib/xmet/bufr.py
View file

@ -82,8 +82,9 @@ class BUFRSounding(Sounding):
DataAccessLayer.changeEDEXHost(BUFRSounding.EDEX_HOST)
@staticmethod
def latest(station: str) -> Self:
def create_request(station: str):
request = DataAccessLayer.newDataRequest()
request.setLocationNames(station)
request.setDatatype(BUFRSounding.BUFR_TYPE)
request.setParameters('staElev', 'staName')
@ -91,21 +92,20 @@ class BUFRSounding(Sounding):
request.getParameters().extend(BUFRSounding.BUFR_PARAMS_SIGT)
request.getParameters().extend(BUFRSounding.BUFR_PARAMS_SIGW)
request.setLocationNames(station)
datatimes = DataAccessLayer.getAvailableTimes(request)
response = DataAccessLayer.getGeometryData(request,
times=datatimes[-1].validPeriod)
return request
@staticmethod
def request_sounding(request, period) -> Self:
sounding = BUFRSounding()
response = DataAccessLayer.getGeometryData(request, times=period)
geom = response[0].getGeometry()
dt = response[0].getDataTime()
epoch = dt.getRefTime().time / 1000.0
timestamp = datetime.datetime.fromtimestamp(epoch, datetime.UTC)
sounding.station = station
sounding.station = response[0].getLocationName()
sounding.data_source_pressure = 'UCAR'
sounding.data_source_other = 'UCAR'
sounding.location = shapely.Point(geom.x, geom.y)
@ -147,3 +147,11 @@ class BUFRSounding(Sounding):
sounding.samples.append(sounding.samples_by_height[height])
return sounding
@staticmethod
def latest(station: str) -> Self:
request = BUFRSounding.create_request(station)
datatimes = DataAccessLayer.getAvailableTimes(request)
return BUFRSounding.request_sounding(request,
datatimes[-1].validPeriod)