Initial commit of lib/xmet/bufr.py

lib/xmet/bufr.py

@@ -0,0 +1,57 @@
+from awips.dataaccess import DataAccessLayer
+
+from xmet.sounding import Sounding, SoundingSample
+from xmet.units    import celsius
+
+class BUFRSounding():
+    EDEX_HOST = 'edex-cloud.unidata.ucar.edu'
+
+    BUFR_TYPE        = 'bufrua'
+    BUFR_PARAMS_MAN  = set(['prMan', 'htMan', 'wdMan', 'wsMan'])
+    BUFR_PARAMS_SIGT = set(['prSigT', 'htSigT', 'tpSigT', 'tdSigT'])
+    BUFR_PARAMS_SIGW = set(['prSigW', 'htSigW', 'wdSigW', 'wsSigW'])
+
+    @staticmethod
+    def init():
+        DataAccessLayer.changeEDEXHost(BUFRSounding.EDEX_HOST)
+
+    @staticmethod
+    def latest(station: str) -> Sounding:
+        request = DataAccessLayer.newDataRequest()
+        request.setDatatype(BUFRSounding.BUFR_TYPE)
+        request.setParameters('staElev', 'staName')
+
+        request.getParameters().extend(BUFRSounding.BUFR_PARAMS_MAN)
+        request.getParameters().extend(BUFRSounding.BUFR_PARAMS_SIGT)
+        request.getParameters().extend(BUFRSounding.BUFR_PARAMS_SIGW)
+
+        request.setLocationName(station)
+
+        datatimes = DataAccessLayer.getAvailableTimes(request)
+
+        response = DataAccessLayer.getGeometryData(request,
+                                                   times=datatimes[-1].validPeriod)
+
+        sounding = Sounding()
+
+        for item in response:
+            params = item.getParameters()
+
+            if set(params) & BUFRSounding.BUFR_PARAMS_MAN:
+                sample = sounding.sample(item.getNumber('prMan') / 100.0)
+                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)
+                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)
+                sample.height     = item.getNumber('htSigW')
+                sample.wind_speed = item.getNumber('wsSigW')
+                sample.wind_dir   = item.getNumber('wdSigW')
+
+        return sounding