Merge master into wunderground

aosstower/level_00/influxdb.py

@@ -15,6 +15,7 @@ from metobscommon.util import calc
 from metobscommon.util.nc import calculate_wind_gust
 import numpy as np
 import pandas as pd
+import warnings
 
 LOG = logging.getLogger(__name__)
 # map station name to InfluxDB tags
@@ -54,53 +55,30 @@ SYMBOLS = list(SYMBOL_CONVERSIONS.values())
 
 
 class Updater(object):
-    """https://feedback.weather.com/customer/en/portal/articles/2924682-pws-upload-protocol?b_id=17298"""
+    """Append weather record (taken as a dict) and do averages when enough data is ready.
+
+    At least 12 minutes of data is required to do averaging: 10 minutes of wind gusts,
+    and wind gusts need 2 minutes of data to calculate.
+
+    This class is created once at startup and calls rolling_average every time new data is available which tries to do
+    averaging every 5 minutes of data added.
+    """
     def __init__(self):
         # Counter that returns rolling average every 5 minutes. independent of
-        # self.data since its length can be 144 == 12 minutes.
+        # self.data since self.data length can be 144 == 12 minutes.
         self.counter = 0
         self.data = {}
 
     def rolling_average(self, record):
-        KNOTS_9 = calc.knots_to_mps(9.)
-        KNOTS_5 = calc.knots_to_mps(5.)
         self.counter += 1
-        database = {k: schema.database_dict[k] for k in schema.met_vars}
         # Add new data to dict.
         for key in record:
             self.data.setdefault(key, []).append(record[key])
-        # 4:50 and 4:45. http://metobs.ssec.wisc.edu/aoss/tower/cgi-bin/data_data.py?&separator=,
-        # &symbols=p:t:td:dir:spd:flux:accum_precip:rh&begin=-00:05:00&end=&interval=
         # If 5 minutes of data are ready, average current data in dict (up to 15 minutes until data is
         # dropped for the first time, then up to 12 minutes thereafter are held). 60 * 5 seconds = 5 minutes.
         if self.counter % 60 == 0:
-            # Appending to a DataFrame is slow. Instead add to a dict in chunks and pass it to the DataFrame.
-            frame = pd.DataFrame(self.data)
-            frame.set_index('timestamp', inplace=True)
-            frame.mask(frame == -99999., inplace=True)
-            frame.fillna(value=np.nan, inplace=True)
-            # Add wind direction components so we can average wind direction properly
-            frame['wind_east'], frame['wind_north'], _ = calc.wind_vector_components(frame['wind_speed'],
-                                                                                     frame['wind_dir'])
-            frame['wind_dir'] = calc.wind_vector_degrees(frame['wind_east'], frame['wind_north'])
-
-            if 'air_temp' in frame and 'rh' in frame and ('dewpoint' in database or 'dewpoint_mean' in database):
-                LOG.info("'dewpoint' is missing from the input file, will calculate "
-                         "it from air temp and relative humidity")
-                frame['dewpoint'] = calc.dewpoint(frame['air_temp'], frame['rh'])
-
-            # 2 minute rolling average of 5 second data.
-            winds_frame_2m = frame[['wind_speed', 'wind_east', 'wind_north']].rolling('2T').mean()
-            frame['wind_speed_2m'] = winds_frame_2m['wind_speed']
-            frame['wind_dir_2m'] = calc.wind_vector_degrees(winds_frame_2m['wind_east'], winds_frame_2m['wind_north'])
-            # TODO: PEAK_DIR IS THE 5 SEC MAX FROM LAST MINUTE IF 5 KTS OVER LAST 2 MINUTE AVG.
-            # 1 minute rolling peaks
-            wind_peak_1m = frame['wind_speed'].rolling(window='1T', center=False).max()
-            # criteria for a fast wind to be considered a wind gust
-            gust_mask = (winds_frame_2m['wind_speed'] >= KNOTS_9) &\
-                        (wind_peak_1m >= winds_frame_2m['wind_speed'] + KNOTS_5)
-            frame['cur_gust'] = wind_peak_1m.mask(~gust_mask)
-            frame['gust_10m'] = calculate_wind_gust(frame['wind_speed'], winds_frame_2m['wind_speed'])
+            # Appending to a DataFrame is slow. Instead, this adds to a dict in chunks and passes it to the DataFrame.
+            frame = self._calculate_averages()
             # Keep data set within minimum window to improve speed.
             # Wind gusts looks at 10 minute intervals, including the first data point which needs 2 minutes of data
             # before it, totalling 12 minutes. Since data is sent every 5 minutes, at 15+ minutes we should
@@ -111,11 +89,43 @@ class Updater(object):
                     self.data[key] = val[-84:]
                 self.counter -= 60
             else:
-                # Make 10 minute gusts at or before 10 minutes nans because data is insufficient.
+                # Make 10 minute gusts before 12 minutes nans because data is insufficient.
                 frame['gust_10m'].mask(frame['gust_10m'] > -1., inplace=True)
             frame.fillna(np.nan, inplace=True)
             return frame
 
+    def _calculate_averages(self):
+        frame = pd.DataFrame(self.data)
+        KNOTS_9 = calc.knots_to_mps(9.)
+        KNOTS_5 = calc.knots_to_mps(5.)
+        database = {k: schema.database_dict[k] for k in schema.met_vars}
+        frame.set_index('timestamp', inplace=True)
+        frame.mask(frame == -99999., inplace=True)
+        frame.fillna(value=np.nan, inplace=True)
+        # Add wind direction components so we can average wind direction properly
+        frame['wind_east'], frame['wind_north'], _ = calc.wind_vector_components(frame['wind_speed'],
+                                                                                 frame['wind_dir'])
+        frame['wind_dir'] = calc.wind_vector_degrees(frame['wind_east'], frame['wind_north'])
+
+        if 'air_temp' in frame and 'rh' in frame and ('dewpoint' in database or 'dewpoint_mean' in database):
+            LOG.info("'dewpoint' is missing from the input file, will calculate "
+                     "it from air temp and relative humidity")
+            frame['dewpoint'] = calc.dewpoint(frame['air_temp'], frame['rh'])
+
+        # 2 minute rolling average of 5 second data.
+        winds_frame_2m = frame[['wind_speed', 'wind_east', 'wind_north']].rolling('2T').mean()
+        frame['wind_speed_2m'] = winds_frame_2m['wind_speed']
+        frame['wind_dir_2m'] = calc.wind_vector_degrees(winds_frame_2m['wind_east'], winds_frame_2m['wind_north'])
+        # TODO: PEAK_DIR IS THE 5 SEC MAX FROM LAST MINUTE IF 5 KTS OVER LAST 2 MINUTE AVG.
+        # 1 minute rolling peaks
+        wind_peak_1m = frame['wind_speed'].rolling(window='1T', center=False).max()
+        # criteria for a fast wind to be considered a wind gust
+        gust_mask = (winds_frame_2m['wind_speed'] >= KNOTS_9) &\
+                    (wind_peak_1m >= winds_frame_2m['wind_speed'] + KNOTS_5)
+        frame['cur_gust'] = wind_peak_1m.mask(~gust_mask)
+        frame['gust_10m'] = calculate_wind_gust(frame['wind_speed'], winds_frame_2m['wind_speed'])
+        return frame
+
 
 def convert_to_influx_frame(record_gen, symbols, debug=False):
     for idx, record in enumerate(record_gen):
@@ -127,6 +137,56 @@ def convert_to_influx_frame(record_gen, symbols, debug=False):
         yield record
 
 
+def construct_url(data):
+    return ('http://weatherstation.wunderground.com/weatherstation/updateweatherstation.php?'
+            'ID={ID}&'
+            'PASSWORD={PASSWORD}&'
+            'action=updateraw&'
+            'dateutc={dateutc}&'
+            'winddir={winddir}&'
+            'winddir_avg2m={winddir_avg2m}&'
+            'windspeedmph={windspeedmph}&'
+            'windspdmph_avg2m={windspdmph_avg2m}&'
+            'windgustmph={windgustmph}&'
+            'windgustmph_10m={windgustmph_10m}&'
+            'humidity={humidity}&'
+            'tempf={tempf}&'
+            'baromin={baromin}&'
+            'dewptf={dewptf}&'
+            'solarradiation={solarradiation}&'
+            'rainin={rainin}&'
+            'dailyrainin={dailyrainin}&'
+            'softwaretype=SSEC-RIG').format(**data)
+
+
+def get_url_data(avg, wu_id, wu_pw):
+    # Information on what paramaters that can be sent:
+    # https://feedback.weather.com/customer/en/portal/articles/2924682-pws-upload-protocol?b_id=17298
+    # For timestamp, want YYYY-MM-DD+hh:mm:ss of last dataset that was averaged, rounded up to nearest minute.
+    timestamp = avg.index[-1].round('1T').isoformat('+')
+    wind_dir = avg['wind_dir'][-1]
+    wind_dir_2m = avg['wind_dir_2m'][-1]
+    # Converts from m/s to mph.
+    wind_speed = avg['wind_speed'][-1] * 2.23694
+    wind_speed_2m = avg['wind_speed_2m'][-1] * 2.23694
+    cur_gust = avg['cur_gust'][-1] * 2.23694
+    gust_10m = avg['gust_10m'][-1] * 2.23694
+    rel_hum = avg['rel_hum'][-1]
+    # Converts degrees Celsius to degrees Fahrenheit
+    air_temp = avg['air_temp'][-1] * 9. / 5. + 32.
+    # hpa to barometric pressure inches
+    pressure = avg['pressure'][-1] * 0.02952998016471232
+    # degrees Celcus to degrees Fahrenheit.
+    dewpoint = avg['dewpoint'][-1] * 9. / 5. + 32.
+    solar_flux = avg['solar_flux'][-1]
+    precip = avg['precip'][-1]
+    accum_precip = avg['accum_precip'][-1]
+    return {'ID': wu_id, 'PASSWORD': wu_pw, 'dateutc': timestamp, 'winddir': wind_dir, 'winddir_avg2m': wind_dir_2m,
+            'windspeedmph': wind_speed, 'windspdmph_avg2m': wind_speed_2m, 'windgustmph': cur_gust,
+            'windgustmph_10m': gust_10m, 'humidity': rel_hum, 'tempf': air_temp, 'baromin': pressure,
+            'dewptf': dewpoint, 'solarradiation': solar_flux, 'rainin': precip, 'dailyrainin': accum_precip}
+
+
 def main():
     import argparse
     parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.ArgumentDefaultsHelpFormatter)
@@ -194,6 +254,7 @@ def main():
             influxdb.insert(lines, host=args.host, port=args.port, dbname=args.dbname)
             # Record is in a list of size 1, but want just the record.
             avg = updater.rolling_average(record[0])
+<<<<<<< HEAD
             # Once every 5 minutes: 0 through 295 seconds inclusive.
             if avg is not None:
                 # For timestamp, want YYYY-MM-DD+hh:mm:ss of last dataset that was averaged, rounded up to nearest minute.
@@ -244,6 +305,16 @@ def main():
                     resp = requests.post(url)
                     if resp != 'success':
                         raise ValueError('Data not received.')
+=======
+            # Once every 5 minutes: 0 through 295 seconds inclusive in 5 second intervals.
+            if avg is not None:
+                url = construct_url(get_url_data(avg, args.wu_id, wu_pw))
+                if wu_pw and args.ldmp:
+                    resp = requests.post(url)
+                    if resp.status_code != 200:
+                        warnings.warn('Data failed to upload to {0} with status code {1}: {2}'.format(
+                            url, resp.status_code, resp.text))
+>>>>>>> master
 
             if args.sleep_interval:
                 time.sleep(args.sleep_interval)