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