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.1.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         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         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['SNWD'] = 'hS'
 320         iemap['PRES'] = 'baro'
 321         iemap['RHUM'] = 'rH'
 322         iemap['WSPD'] = 'windSpeedNum'
 323         iemap['WDIR'] = 'windDirectionNum'
 324         # unsupported by NRCS:
 325         # windGustSpeedNum
 326     elif provider == 'mesowest':
 327         iemap['precip_accum'] = 'precipitationGauge'
 328         iemap['air_temp'] = 'tempPres'
 329         iemap['snow_depth'] = 'hS'
 330         iemap['pressure'] = 'baro'
 331         iemap['relative_humidity'] = 'rH'
 332         iemap['wind_speed'] = 'windSpeedNum'
 333         iemap['wind_direction'] = 'windDirectionNum'
 334         iemap['wind_gust'] = 'windGustSpeedNum'
 335
 336     return iemap
 337
 338 # provider-specific operations
 339 def get_nrcs_data(begin, end, station):
 340     """get the data we're after from the NRCS WSDL"""
 341     transport = zeep.transports.Transport(cache=zeep.cache.SqliteCache())
 342     client = zeep.Client(wsdl=station['source'], transport=transport)
 343     remote_data = {}
 344
 345     for element_cd in station['desired_data']:
 346         time_element = time.time()
 347
 348         # get the last three hours of data for this elementCd/element_cd
 349         tmp = client.service.getHourlyData(
 350             stationTriplets=[station['station_id']],
 351             elementCd=element_cd,
 352             ordinal=1,
 353             beginDate=begin,
 354             endDate=end)
 355
 356         LOG.info("Time to get NRCS elementCd '%s': %.3f sec", element_cd,
 357                  time.time() - time_element)
 358
 359         values = tmp[0]['values']
 360
 361         # sort and isolate the most recent
 362         #
 363         # NOTE: we do this because sometimes there are gaps in hourly data
 364         #       in NRCS; yes, we may end up with slightly inaccurate data,
 365         #       so perhaps this decision will be re-evaluated in the future
 366         if values:
 367             ordered = sorted(values, key=lambda t: t['dateTime'], reverse=True)
 368             remote_data[element_cd] = ordered[0]['value']
 369         else:
 370             remote_data[element_cd] = None
 371
 372     return remote_data
 373
 374 def get_mesowest_data(begin, end, station):
 375     """get the data we're after from the MesoWest/Synoptic API"""
 376     remote_data = {}
 377
 378     # massage begin/end date format
 379     begin_date_str = begin.strftime('%Y%m%d%H%M')
 380     end_date_str = end.strftime('%Y%m%d%H%M')
 381
 382     # construct final, completed API URL
 383     api_req_url = station['source'] + '&start=' + begin_date_str + '&end=' + end_date_str
 384     req = requests.get(api_req_url)
 385
 386     try:
 387         json = req.json()
 388     except ValueError:
 389         LOG.error("Bad JSON in MesoWest response")
 390         sys.exit(1)
 391
 392     try:
 393         observations = json['STATION'][0]['OBSERVATIONS']
 394     except ValueError:
 395         LOG.error("Bad JSON in MesoWest response")
 396         sys.exit(1)
 397
 398     pos = len(observations['date_time']) - 1
 399
 400     for element_cd in station['desired_data'].split(','):
 401         # sort and isolate the most recent, see note above in NRCS for how and
 402         # why this is done
 403         #
 404         # NOTE: Unlike in the NRCS case, the MesoWest API response contains all
 405         #       data (whereas with NRCS, we have to make a separate request for
 406         #       each element we want). This is nice for network efficiency but
 407         #       it means we have to handle this part differently for each.
 408         #
 409         # NOTE: Also unlike NRCS, MesoWest provides more granular data; NRCS
 410         #       provides hourly data, but MesoWest can often provide data every
 411         #       10 minutes -- though this provides more opportunity for
 412         #       irregularities
 413
 414         # we may not have the data at all
 415         key_name = element_cd + '_set_1'
 416         if key_name in observations:
 417             if observations[key_name][pos]:
 418                 remote_data[element_cd] = observations[key_name][pos]
 419             else:
 420                 remote_data[element_cd] = None
 421         else:
 422             remote_data[element_cd] = None
 423
 424     return remote_data
 425
 426 def switch_units_to_metric(data_map, mapping):
 427     """replace units with metric counterparts"""
 428
 429     # NOTE: to update this, use the fmap<->final_data mapping laid out
 430     #       in setup_infoex_fields_mapping ()
 431     #
 432     # NOTE: this only 'works' with MesoWest for now, as the MesoWest API
 433     #       itself handles the unit conversion; in the future, we will also
 434     #       support NRCS unit conversion, but this must be done by this
 435     #       program.
 436     data_map[mapping['tempPresUnit']] = 'C'
 437     data_map[mapping['hsUnit']] = 'm'
 438     data_map[mapping['windSpeedUnit']] = 'm/s'
 439     data_map[mapping['windGustSpeedNumUnit']] = 'm/s'
 440
 441     return data_map
 442
 443 # CSV operations
 444 def write_local_csv(path_to_file, data):
 445     """Write the specified CSV file to disk"""
 446     with open(path_to_file, 'w') as file_object:
 447         # The requirement is that empty values are represented in the CSV
 448         # file as "", csv.QUOTE_NONNUMERIC achieves that
 449         LOG.debug("writing CSV file '%s'", path_to_file)
 450         writer = csv.writer(file_object, quoting=csv.QUOTE_NONNUMERIC)
 451         writer.writerow(data)
 452         file_object.close()
 453     return True
 454
 455 def upload_csv(path_to_file, infoex_data):
 456     """Upload the specified CSV file to InfoEx FTP and remove the file"""
 457     with open(path_to_file, 'rb') as file_object:
 458         LOG.debug("uploading FTP file '%s'", infoex_data['host'])
 459         ftp = FTP(infoex_data['host'], infoex_data['uuid'],
 460                   infoex_data['api_key'])
 461         ftp.storlines('STOR ' + path_to_file, file_object)
 462         ftp.close()
 463         file_object.close()
 464     os.remove(path_to_file)
 465
 466 # other miscellaneous routines
 467 def setup_time_values():
 468     """establish time bounds of data request(s)"""
 469     # floor time to nearest hour
 470     date_time = datetime.datetime.now()
 471     end_date = date_time - datetime.timedelta(minutes=date_time.minute % 60,
 472                                               seconds=date_time.second,
 473                                               microseconds=date_time.microsecond)
 474     begin_date = end_date - datetime.timedelta(hours=3)
 475     return (begin_date, end_date)
 476
 477 if __name__ == "__main__":
 478     sys.exit(main())