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