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