infoex-autowx.py

   1 #!/usr/bin/env python
   2 # -*- coding: utf-8 -*-
   3
   4 """
   5 InfoEx <-> NRCS/MesoWest Auto Wx implementation
   6 Alexander Vasarab
   7 Wylark Mountaineering LLC
   8
   9 This program fetches data from either an NRCS SNOTEL site or MesoWest
  10 weather station and pushes it to InfoEx using the new automated weather
  11 system implementation.
  12
  13 It is designed to be run hourly, and it asks for the last three hours
  14 of data of each desired type, and selects the most recent one. This
  15 lends some resiliency to the process and helps ensure that we have a
  16 value to send, but it can lead to somewhat inconsistent/untruthful
  17 data if e.g. the HS is from the last hour but the tempPres is from two
  18 hours ago because the instrumentation had a hiccup. It's worth
  19 considering if this is a bug or a feature.
  20
  21 For more information, see file: README
  22 For licensing, see file: LICENSE
  23 """
  24
  25 import configparser
  26 import csv
  27 import datetime
  28 import logging
  29 import os
  30 import sys
  31 import time
  32
  33 from ftplib import FTP
  34 from argparse import ArgumentParser
  35
  36 import requests
  37
  38 import zeep
  39 import zeep.cache
  40 import zeep.transports
  41
  42 __version__ = '2.2.0'
  43
  44 LOG = logging.getLogger(__name__)
  45 LOG.setLevel(logging.NOTSET)
  46
  47 def get_parser():
  48     """Return OptionParser for this program"""
  49     parser = ArgumentParser()
  50
  51     parser.add_argument("--version",
  52                         action="version",
  53                         version=__version__)
  54
  55     parser.add_argument("--config",
  56                         dest="config",
  57                         metavar="FILE",
  58                         help="location of config file")
  59
  60     parser.add_argument("--log-level",
  61                         dest="log_level",
  62                         default=None,
  63                         help="set the log level (debug, info, warning)")
  64
  65     parser.add_argument("--dry-run",
  66                         action="store_true",
  67                         dest="dry_run",
  68                         default=False,
  69                         help="fetch data but don't upload to InfoEx")
  70
  71     return parser
  72
  73 def setup_config(config):
  74     """Setup config variable based on values specified in the ini file"""
  75     try:
  76         infoex = {
  77             'host': config['infoex']['host'],
  78             'uuid': config['infoex']['uuid'],
  79             'api_key': config['infoex']['api_key'],
  80             'csv_filename': config['infoex']['csv_filename'],
  81             'location_uuid': config['infoex']['location_uuid'],
  82             'wx_data': {}, # placeholder key, values to come later
  83         }
  84
  85         station = dict()
  86         station['provider'] = config['station']['type']
  87
  88         if station['provider'] not in ['nrcs', 'mesowest']:
  89             print("Please specify either nrcs or mesowest as the station type.")
  90             sys.exit(1)
  91
  92         if station['provider'] == 'nrcs':
  93             station['source'] = 'https://www.wcc.nrcs.usda.gov/awdbWebService/services?WSDL'
  94             station['station_id'] = config['station']['station_id']
  95             station['desired_data'] = config['station']['desired_data'].split(',')
  96
  97             # XXX: For NRCS, we're manually overriding units for now! Once
  98             #      unit conversion is supported for NRCS, REMOVE THIS!
  99             if 'units' not in station:
 100                 station['units'] = 'imperial'
 101
 102         if station['provider'] == 'mesowest':
 103             station['source'] = 'https://api.synopticdata.com/v2/stations/timeseries'
 104             station['station_id'] = config['station']['station_id']
 105             station['units'] = config['station']['units']
 106             station['desired_data'] = config['station']['desired_data']
 107
 108             # construct full API URL (sans start/end time, added later)
 109             station['source'] = station['source'] + '?token=' + \
 110                                 config['station']['token'] + \
 111                                 '&within=60&units=' + station['units'] + \
 112                                 '&stid=' + station['station_id'] + \
 113                                 '&vars=' + station['desired_data']
 114
 115     except KeyError as err:
 116         LOG.critical("%s not defined in configuration file", err)
 117         sys.exit(1)
 118
 119     # all sections/values present in config file, final sanity check
 120     try:
 121         for key in config.sections():
 122             for subkey in config[key]:
 123                 if not config[key][subkey]:
 124                     raise ValueError
 125     except ValueError:
 126         LOG.critical("Config value '%s.%s' is empty", key, subkey)
 127         sys.exit(1)
 128
 129     return (infoex, station)
 130
 131 def setup_logging(log_level):
 132     """Setup our logging infrastructure"""
 133     try:
 134         from systemd.journal import JournalHandler
 135         LOG.addHandler(JournalHandler())
 136     except ImportError:
 137         ## fallback to syslog
 138         #import logging.handlers
 139         #LOG.addHandler(logging.handlers.SysLogHandler())
 140         # fallback to stdout
 141         handler = logging.StreamHandler(sys.stdout)
 142         LOG.addHandler(handler)
 143
 144     # ugly, but passable
 145     if log_level in [None, 'debug', 'info', 'warning']:
 146         if log_level == 'debug':
 147             LOG.setLevel(logging.DEBUG)
 148         elif log_level == 'info':
 149             LOG.setLevel(logging.INFO)
 150         elif log_level == 'warning':
 151             LOG.setLevel(logging.WARNING)
 152         else:
 153             LOG.setLevel(logging.NOTSET)
 154     else:
 155         return False
 156
 157     return True
 158
 159 def main():
 160     """Main routine: sort through args, decide what to do, then do it"""
 161     parser = get_parser()
 162     options = parser.parse_args()
 163
 164     config = configparser.ConfigParser(allow_no_value=False)
 165
 166     if not options.config:
 167         parser.print_help()
 168         print("\nPlease specify a configuration file via --config.")
 169         sys.exit(1)
 170
 171     config.read(options.config)
 172
 173     if not setup_logging(options.log_level):
 174         parser.print_help()
 175         print("\nPlease select an appropriate log level or remove the switch (--log-level).")
 176         sys.exit(1)
 177
 178     (infoex, station) = setup_config(config)
 179
 180     LOG.debug('Config parsed, starting up')
 181
 182     # create mappings
 183     (fmap, final_data) = setup_infoex_fields_mapping(infoex['location_uuid'])
 184     iemap = setup_infoex_counterparts_mapping(station['provider'])
 185
 186     # override units if user selected metric
 187     if station['units'] == 'metric':
 188         final_data = switch_units_to_metric(final_data, fmap)
 189
 190     (begin_date, end_date) = setup_time_values()
 191
 192     # get the data
 193     LOG.debug("Getting %s data from %s to %s", str(station['desired_data']),
 194               str(begin_date), str(end_date))
 195
 196     time_all_elements = time.time()
 197
 198     # get the data
 199     if station['provider'] == 'nrcs':
 200         infoex['wx_data'] = get_nrcs_data(begin_date, end_date, station)
 201     elif station['provider'] == 'mesowest':
 202         infoex['wx_data'] = get_mesowest_data(begin_date, end_date,
 203                                               station)
 204
 205     LOG.info("Time taken to get all data : %.3f sec", time.time() -
 206              time_all_elements)
 207
 208     LOG.debug("infoex[wx_data]: %s", str(infoex['wx_data']))
 209
 210     # Now we only need to add in what we want to change thanks to that
 211     # abomination of a variable declaration earlier
 212     final_data[fmap['Location UUID']] = infoex['location_uuid']
 213     final_data[fmap['obDate']] = end_date.strftime('%m/%d/%Y')
 214     final_data[fmap['obTime']] = end_date.strftime('%H:%M')
 215
 216     for element_cd in infoex['wx_data']:
 217         if element_cd not in iemap:
 218             LOG.warning("BAD KEY wx_data['%s']", element_cd)
 219             continue
 220
 221         # Massage precision of certain values to fit InfoEx's
 222         # expectations
 223         #
 224         # 0 decimal places: wind speed, wind direction, wind gust, snow depth
 225         # 1 decimal place:  air temp, baro
 226         # Avoid transforming None values
 227         if infoex['wx_data'][element_cd] is None:
 228             continue
 229         elif element_cd in ['wind_speed', 'WSPD', 'wind_direction',
 230                             'WDIR', 'wind_gust', 'SNWD', 'snow_depth']:
 231             infoex['wx_data'][element_cd] = round(infoex['wx_data'][element_cd])
 232         elif element_cd in ['TOBS', 'air_temp', 'PRES', 'pressure']:
 233             infoex['wx_data'][element_cd] = round(infoex['wx_data'][element_cd], 1)
 234
 235         # CONSIDER: Casting every value to Float() -- need to investigate if
 236         #           any possible elementCds we may want are any other data
 237         #           type than float.
 238         #
 239         #           Another possibility is to query the API with
 240         #           getStationElements and temporarily store the
 241         #           storedUnitCd. But that's pretty network-intensive and
 242         #           may not even be worth it if there's only e.g. one or two
 243         #           exceptions to any otherwise uniformly Float value set.
 244         final_data[fmap[iemap[element_cd]]] = infoex['wx_data'][element_cd]
 245
 246     LOG.debug("final_data: %s", str(final_data))
 247
 248     if not write_local_csv(infoex['csv_filename'], final_data):
 249         LOG.warning('Could not write local CSV file: %s',
 250                     infoex['csv_filename'])
 251         return 1
 252
 253     if not options.dry_run:
 254         upload_csv(infoex['csv_filename'], infoex)
 255
 256     LOG.debug('DONE')
 257     return 0
 258
 259 # data structure operations
 260 def setup_infoex_fields_mapping(location_uuid):
 261     """
 262     Create a mapping of InfoEx fields to the local data's indexing scheme.
 263
 264     INFOEX FIELDS
 265
 266     This won't earn style points in Python, but here we establish a couple
 267     of helpful mappings variables. The reason this is helpful is that the
 268     end result is simply an ordered set, the CSV file. But we still may
 269     want to manipulate the values arbitrarily before writing that file.
 270
 271     Also note that the current Auto Wx InfoEx documentation shows these
 272     keys in a graphical table with the "index" beginning at 1, but here we
 273     sanely index beginning at 0.
 274     """
 275     # pylint: disable=too-many-statements,multiple-statements,bad-whitespace
 276     fmap = {}                           ; final_data     = [None] * 29
 277     fmap['Location UUID'] = 0           ; final_data[0]  = location_uuid
 278     fmap['obDate'] = 1                  ; final_data[1]  = None
 279     fmap['obTime'] = 2                  ; final_data[2]  = None
 280     fmap['timeZone'] = 3                ; final_data[3]  = 'Pacific'
 281     fmap['tempMaxHour'] = 4             ; final_data[4]  = None
 282     fmap['tempMaxHourUnit'] = 5         ; final_data[5]  = 'F'
 283     fmap['tempMinHour'] = 6             ; final_data[6]  = None
 284     fmap['tempMinHourUnit'] = 7         ; final_data[7]  = 'F'
 285     fmap['tempPres'] = 8                ; final_data[8]  = None
 286     fmap['tempPresUnit'] = 9            ; final_data[9]  = 'F'
 287     fmap['precipitationGauge'] = 10     ; final_data[10] = None
 288     fmap['precipitationGaugeUnit'] = 11 ; final_data[11] = 'in'
 289     fmap['windSpeedNum'] = 12           ; final_data[12] = None
 290     fmap['windSpeedUnit'] = 13          ; final_data[13] = 'mph'
 291     fmap['windDirectionNum'] = 14       ; final_data[14] = None
 292     fmap['hS'] = 15                     ; final_data[15] = None
 293     fmap['hsUnit'] = 16                 ; final_data[16] = 'in'
 294     fmap['baro'] = 17                   ; final_data[17] = None
 295     fmap['baroUnit'] = 18               ; final_data[18] = 'inHg'
 296     fmap['rH'] = 19                     ; final_data[19] = None
 297     fmap['windGustSpeedNum'] = 20       ; final_data[20] = None
 298     fmap['windGustSpeedNumUnit'] = 21   ; final_data[21] = 'mph'
 299     fmap['windGustDirNum'] = 22         ; final_data[22] = None
 300     fmap['dewPoint'] = 23               ; final_data[23] = None
 301     fmap['dewPointUnit'] = 24           ; final_data[24] = 'F'
 302     fmap['hn24Auto'] = 25               ; final_data[25] = None
 303     fmap['hn24AutoUnit'] = 26           ; final_data[26] = 'in'
 304     fmap['hstAuto'] = 27                ; final_data[27] = None
 305     fmap['hstAutoUnit'] = 28            ; final_data[28] = 'in'
 306
 307     return (fmap, final_data)
 308
 309 def setup_infoex_counterparts_mapping(provider):
 310     """
 311     Create a mapping of the NRCS/MesoWest fields that this program supports to
 312     their InfoEx counterparts
 313     """
 314     iemap = {}
 315
 316     if provider == 'nrcs':
 317         iemap['PREC'] = 'precipitationGauge'
 318         iemap['TOBS'] = 'tempPres'
 319         iemap['TMAX'] = 'tempMaxHour'
 320         iemap['TMIN'] = 'tempMinHour'
 321         iemap['SNWD'] = 'hS'
 322         iemap['PRES'] = 'baro'
 323         iemap['RHUM'] = 'rH'
 324         iemap['WSPD'] = 'windSpeedNum'
 325         iemap['WDIR'] = 'windDirectionNum'
 326         # unsupported by NRCS:
 327         # windGustSpeedNum
 328     elif provider == 'mesowest':
 329         iemap['precip_accum'] = 'precipitationGauge'
 330         iemap['air_temp'] = 'tempPres'
 331         iemap['air_temp_high_24_hour'] = 'tempMaxHour'
 332         iemap['air_temp_low_24_hour'] = 'tempMinHour'
 333         iemap['snow_depth'] = 'hS'
 334         iemap['pressure'] = 'baro'
 335         iemap['relative_humidity'] = 'rH'
 336         iemap['wind_speed'] = 'windSpeedNum'
 337         iemap['wind_direction'] = 'windDirectionNum'
 338         iemap['wind_gust'] = 'windGustSpeedNum'
 339
 340     return iemap
 341
 342 # provider-specific operations
 343 def get_nrcs_data(begin, end, station):
 344     """get the data we're after from the NRCS WSDL"""
 345     transport = zeep.transports.Transport(cache=zeep.cache.SqliteCache())
 346     client = zeep.Client(wsdl=station['source'], transport=transport)
 347     remote_data = {}
 348
 349     for element_cd in station['desired_data']:
 350         time_element = time.time()
 351
 352         # get the last three hours of data for this elementCd/element_cd
 353         tmp = client.service.getHourlyData(
 354             stationTriplets=[station['station_id']],
 355             elementCd=element_cd,
 356             ordinal=1,
 357             beginDate=begin,
 358             endDate=end)
 359
 360         LOG.info("Time to get NRCS elementCd '%s': %.3f sec", element_cd,
 361                  time.time() - time_element)
 362
 363         values = tmp[0]['values']
 364
 365         # sort and isolate the most recent
 366         #
 367         # NOTE: we do this because sometimes there are gaps in hourly data
 368         #       in NRCS; yes, we may end up with slightly inaccurate data,
 369         #       so perhaps this decision will be re-evaluated in the future
 370         if values:
 371             ordered = sorted(values, key=lambda t: t['dateTime'], reverse=True)
 372             remote_data[element_cd] = ordered[0]['value']
 373         else:
 374             remote_data[element_cd] = None
 375
 376     return remote_data
 377
 378 def get_mesowest_data(begin, end, station):
 379     """get the data we're after from the MesoWest/Synoptic API"""
 380     remote_data = {}
 381
 382     # massage begin/end date format
 383     begin_date_str = begin.strftime('%Y%m%d%H%M')
 384     end_date_str = end.strftime('%Y%m%d%H%M')
 385
 386     # construct final, completed API URL
 387     api_req_url = station['source'] + '&start=' + begin_date_str + '&end=' + end_date_str
 388     req = requests.get(api_req_url)
 389
 390     try:
 391         json = req.json()
 392     except ValueError:
 393         LOG.error("Bad JSON in MesoWest response")
 394         sys.exit(1)
 395
 396     try:
 397         observations = json['STATION'][0]['OBSERVATIONS']
 398     except ValueError:
 399         LOG.error("Bad JSON in MesoWest response")
 400         sys.exit(1)
 401
 402     pos = len(observations['date_time']) - 1
 403
 404     for element_cd in station['desired_data'].split(','):
 405         # sort and isolate the most recent, see note above in NRCS for how and
 406         # why this is done
 407         #
 408         # NOTE: Unlike in the NRCS case, the MesoWest API response contains all
 409         #       data (whereas with NRCS, we have to make a separate request for
 410         #       each element we want). This is nice for network efficiency but
 411         #       it means we have to handle this part differently for each.
 412         #
 413         # NOTE: Also unlike NRCS, MesoWest provides more granular data; NRCS
 414         #       provides hourly data, but MesoWest can often provide data every
 415         #       10 minutes -- though this provides more opportunity for
 416         #       irregularities
 417
 418         # we may not have the data at all
 419         key_name = element_cd + '_set_1'
 420         if key_name in observations:
 421             if observations[key_name][pos]:
 422                 remote_data[element_cd] = observations[key_name][pos]
 423             else:
 424                 remote_data[element_cd] = None
 425         else:
 426             remote_data[element_cd] = None
 427
 428     return remote_data
 429
 430 def switch_units_to_metric(data_map, mapping):
 431     """replace units with metric counterparts"""
 432
 433     # NOTE: to update this, use the fmap<->final_data mapping laid out
 434     #       in setup_infoex_fields_mapping ()
 435     #
 436     # NOTE: this only 'works' with MesoWest for now, as the MesoWest API
 437     #       itself handles the unit conversion; in the future, we will also
 438     #       support NRCS unit conversion, but this must be done by this
 439     #       program.
 440     data_map[mapping['tempPresUnit']] = 'C'
 441     data_map[mapping['hsUnit']] = 'm'
 442     data_map[mapping['windSpeedUnit']] = 'm/s'
 443     data_map[mapping['windGustSpeedNumUnit']] = 'm/s'
 444
 445     return data_map
 446
 447 # CSV operations
 448 def write_local_csv(path_to_file, data):
 449     """Write the specified CSV file to disk"""
 450     with open(path_to_file, 'w') as file_object:
 451         # The requirement is that empty values are represented in the CSV
 452         # file as "", csv.QUOTE_NONNUMERIC achieves that
 453         LOG.debug("writing CSV file '%s'", path_to_file)
 454         writer = csv.writer(file_object, quoting=csv.QUOTE_NONNUMERIC)
 455         writer.writerow(data)
 456         file_object.close()
 457     return True
 458
 459 def upload_csv(path_to_file, infoex_data):
 460     """Upload the specified CSV file to InfoEx FTP and remove the file"""
 461     with open(path_to_file, 'rb') as file_object:
 462         LOG.debug("uploading FTP file '%s'", infoex_data['host'])
 463         ftp = FTP(infoex_data['host'], infoex_data['uuid'],
 464                   infoex_data['api_key'])
 465         ftp.storlines('STOR ' + path_to_file, file_object)
 466         ftp.close()
 467         file_object.close()
 468     os.remove(path_to_file)
 469
 470 # other miscellaneous routines
 471 def setup_time_values():
 472     """establish time bounds of data request(s)"""
 473     # floor time to nearest hour
 474     date_time = datetime.datetime.now()
 475     end_date = date_time - datetime.timedelta(minutes=date_time.minute % 60,
 476                                               seconds=date_time.second,
 477                                               microseconds=date_time.microsecond)
 478     begin_date = end_date - datetime.timedelta(hours=3)
 479     return (begin_date, end_date)
 480
 481 if __name__ == "__main__":
 482     sys.exit(main())