weather.py

   1 # Copyright (c) 2006-2010 Jeremy Stanley <fungi@yuggoth.org>. Permission to
   2 # use, copy, modify, and distribute this software is granted under terms
   3 # provided in the LICENSE file distributed with this software.
   4
   5 """Contains various object definitions needed by the weather utility."""
   6
   7 version = "1.5"
   8
   9 class Selections:
  10    """An object to contain selection data."""
  11    def __init__(self):
  12       """Store the config, options and arguments."""
  13       self.config = get_config()
  14       self.options, self.arguments = get_options(self.config)
  15       if self.arguments:
  16          self.arguments = [(x.lower()) for x in self.arguments]
  17       else: self.arguments = [ None ]
  18    def get(self, option, argument=None):
  19       """Retrieve data from the config or options."""
  20       if not argument: return self.options.__dict__[option]
  21       elif not self.config.has_section(argument):
  22          import sys
  23          sys.stderr.write("weather: error: no alias defined for " \
  24             + argument + "\n")
  25          sys.exit(1)
  26       elif self.config.has_option(argument, option):
  27          return self.config.get(argument, option)
  28       else: return self.options.__dict__[option]
  29    def get_bool(self, option, argument=None):
  30       """Get data and coerce to a boolean if necessary."""
  31       return bool(self.get(option, argument))
  32
  33 def bool(data):
  34    """Coerce data to a boolean value."""
  35    if type(data) is str:
  36       if eval(data): return True
  37       else: return False
  38    else:
  39       if data: return True
  40       else: return False
  41
  42 def quote(words):
  43    """Wrap a string in quotes if it contains spaces."""
  44    if words.find(" ") != -1: words = "\"" + words + "\""
  45    return words
  46
  47 def titlecap(words):
  48    """Perform English-language title capitalization."""
  49    words = words.lower().strip()
  50    for separator in [" ", "-", "'"]:
  51       newwords = []
  52       wordlist = words.split(separator)
  53       for word in wordlist:
  54          if word:
  55             newwords.append(word[0].upper() + word[1:])
  56       words = separator.join(newwords)
  57    end = len(words)
  58    for prefix in ["Mac", "Mc"]:
  59       position = 0
  60       offset = len(prefix)
  61       while position < end:
  62          position = words.find(prefix, position)
  63          if position == -1:
  64             position = end
  65          position += offset
  66          import string
  67          if position < end and words[position] in string.letters:
  68             words = words[:position] \
  69                + words[position].upper() \
  70                + words[position+1:]
  71    return words
  72
  73 def filter_units(line, units="imperial"):
  74    """Filter or convert units in a line of text between US/UK and metric."""
  75    import re
  76    # filter lines with both pressures in the form of "X inches (Y hPa)" or
  77    # "X in. Hg (Y hPa)"
  78    dual_p = re.match(
  79       "(.* )(\d*(\.\d+)? (inches|in\. Hg)) \((\d*(\.\d+)? hPa)\)(.*)",
  80       line
  81    )
  82    if dual_p:
  83       preamble, in_hg, i_fr, i_un, hpa, h_fr, trailer = dual_p.groups()
  84       if units == "imperial": line = preamble + in_hg + trailer
  85       elif units == "metric": line = preamble + hpa + trailer
  86    # filter lines with both temperatures in the form of "X F (Y C)"
  87    dual_t = re.match(
  88       "(.* )(\d*(\.\d+)? F) \((\d*(\.\d+)? C)\)(.*)",
  89       line
  90    )
  91    if dual_t:
  92       preamble, fahrenheit, f_fr, celsius, c_fr, trailer = dual_t.groups()
  93       if units == "imperial": line = preamble + fahrenheit + trailer
  94       elif units == "metric": line = preamble + celsius + trailer
  95    # if metric is desired, convert distances in the form of "X mile(s)" to
  96    # "Y kilometer(s)"
  97    if units == "metric":
  98       imperial_d = re.match(
  99          "(.* )(\d+)( mile\(s\))(.*)",
 100          line
 101       )
 102       if imperial_d:
 103          preamble, mi, m_u, trailer = imperial_d.groups()
 104          line = preamble + str(int(round(int(mi)*1.609344))) \
 105             + " kilometer(s)" + trailer
 106    # filter speeds in the form of "X MPH (Y KT)" to just "X MPH"; if metric is
 107    # desired, convert to "Z KPH"
 108    imperial_s = re.match(
 109       "(.* )(\d+)( MPH)( \(\d+ KT\))(.*)",
 110       line
 111    )
 112    if imperial_s:
 113       preamble, mph, m_u, kt, trailer = imperial_s.groups()
 114       if units == "imperial": line = preamble + mph + m_u + trailer
 115       elif units == "metric":
 116          line = preamble + str(int(round(int(mph)*1.609344))) + " KPH" + \
 117             trailer
 118    # if imperial is desired, qualify given forcast temperatures like "X F"; if
 119    # metric is desired, convert to "Y C"
 120    imperial_t = re.match(
 121       "(.* )(High |high |Low |low )(\d+)(\.|,)(.*)",
 122       line
 123    )
 124    if imperial_t:
 125       preamble, parameter, fahrenheit, sep, trailer = imperial_t.groups()
 126       if units == "imperial":
 127          line = preamble + parameter + fahrenheit + " F" + sep + trailer
 128       elif units == "metric":
 129          line = preamble + parameter \
 130             + str(int(round((int(fahrenheit)-32)*5/9))) + " C" + sep + trailer
 131    # hand off the resulting line
 132    return line
 133
 134 def sorted(data):
 135    """Return a sorted copy of a list."""
 136    new_copy = data[:]
 137    new_copy.sort()
 138    return new_copy
 139
 140 def get_url(url, ignore_fail=False):
 141    """Return a string containing the results of a URL GET."""
 142    import urllib2
 143    try: return urllib2.urlopen(url).read()
 144    except urllib2.URLError:
 145       if ignore_fail: return ""
 146       else:
 147          import sys, traceback
 148          sys.stderr.write("weather: error: failed to retrieve\n   " \
 149             + url + "\n   " + \
 150             traceback.format_exception_only(sys.exc_type, sys.exc_value)[0])
 151          sys.exit(1)
 152
 153 def get_metar(
 154    id,
 155    verbose=False,
 156    quiet=False,
 157    headers=None,
 158    murl=None,
 159    imperial=False,
 160    metric=False
 161 ):
 162    """Return a summarized METAR for the specified station."""
 163    if not id:
 164       import sys
 165       sys.stderr.write("weather: error: id required for conditions\n")
 166       sys.exit(1)
 167    if not murl:
 168       murl = \
 169          "http://weather.noaa.gov/pub/data/observations/metar/decoded/%ID%.TXT"
 170    murl = murl.replace("%ID%", id.upper())
 171    murl = murl.replace("%Id%", id.capitalize())
 172    murl = murl.replace("%iD%", id)
 173    murl = murl.replace("%id%", id.lower())
 174    murl = murl.replace(" ", "_")
 175    metar = get_url(murl)
 176    if verbose: return metar
 177    else:
 178       lines = metar.split("\n")
 179       if not headers:
 180          headers = \
 181             "relative_humidity," \
 182             + "precipitation_last_hour," \
 183             + "sky conditions," \
 184             + "temperature," \
 185             + "weather," \
 186             + "wind"
 187       headerlist = headers.lower().replace("_"," ").split(",")
 188       output = []
 189       if not quiet:
 190          title = "Current conditions at %s"
 191          place = lines[0].split(", ")
 192          if len(place) > 1:
 193             place = "%s, %s (%s)" % (titlecap(place[0]), place[1], id.upper())
 194          else: place = id.upper()
 195          output.append(title%place)
 196          output.append("Last updated " + lines[1])
 197       for header in headerlist:
 198          for line in lines:
 199             if line.lower().startswith(header + ":"):
 200                if line.endswith(":0") or line.endswith(":1"):
 201                   line = line[:-2]
 202                if imperial: line = filter_units(line, units="imperial")
 203                elif metric: line = filter_units(line, units="metric")
 204                if quiet: output.append(line)
 205                else: output.append("   " + line)
 206       return "\n".join(output)
 207
 208 def get_alert(
 209    zone,
 210    verbose=False,
 211    quiet=False,
 212    atype=None,
 213    aurl=None,
 214    imperial=False,
 215    metric=False
 216 ):
 217    """Return alert notice for the specified zone and type."""
 218    if not zone:
 219       import sys
 220       sys.stderr.write("weather: error: zone required for alerts\n")
 221       sys.exit(1)
 222    if not atype: atype = "severe_weather_stmt"
 223    if not aurl:
 224       aurl = \
 225          "http://weather.noaa.gov/pub/data/watches_warnings/%atype%/%zone%.txt"
 226    aurl = aurl.replace("%ATYPE%", atype.upper())
 227    aurl = aurl.replace("%Atype%", atype.capitalize())
 228    aurl = aurl.replace("%atypE%", atype)
 229    aurl = aurl.replace("%atype%", atype.lower())
 230    aurl = aurl.replace("%ZONE%", zone.upper())
 231    aurl = aurl.replace("%Zone%", zone.capitalize())
 232    aurl = aurl.replace("%zonE%", zone)
 233    aurl = aurl.replace("%zone%", zone.lower())
 234    aurl = aurl.replace(" ", "_")
 235    alert = get_url(aurl, ignore_fail=True).strip()
 236    if alert:
 237       if verbose: return alert
 238       else:
 239          lines = alert.split("\n")
 240          muted = True
 241          import calendar, re, time
 242          valid_time = time.strftime("%Y%m%d%H%M")
 243          #if not quiet: output = [ lines[3], lines[5] ]
 244          #if not quiet: output = [ lines[8], lines[10] ]
 245          #else: output = []
 246          output = []
 247          for line in lines:
 248             if line.startswith("Expires:") and "Expires:"+valid_time > line:
 249                return ""
 250             if muted and line.find("...") != -1:
 251                muted = False
 252             if line == "$$" \
 253                or line.startswith("LAT...LON") \
 254                or line.startswith("TIME...MOT...LOC"):
 255                muted = True
 256             if line and not (
 257                muted \
 258                or line == "&&"
 259                or re.match("^/.*/$", line) \
 260                or re.match("^"+zone.split("/")[1][:3].upper()+".*", line)
 261             ):
 262                if quiet: output.append(line)
 263                else: output.append("   " + line)
 264          return "\n".join(output)
 265
 266 def get_forecast(
 267    city,
 268    st,
 269    verbose=False,
 270    quiet=False,
 271    flines="0",
 272    furl=None,
 273    imperial=False,
 274    metric=False
 275 ):
 276    """Return the forecast for a specified city/st combination."""
 277    if not city or not st:
 278       import sys
 279       sys.stderr.write("weather: error: city and st required for forecast\n")
 280       sys.exit(1)
 281    if not furl:
 282       furl = "http://weather.noaa.gov/pub/data/forecasts/city/%st%/%city%.txt"
 283    furl = furl.replace("%CITY%", city.upper())
 284    furl = furl.replace("%City%", city.capitalize())
 285    furl = furl.replace("%citY%", city)
 286    furl = furl.replace("%city%", city.lower())
 287    furl = furl.replace("%ST%", st.upper())
 288    furl = furl.replace("%St%", st.capitalize())
 289    furl = furl.replace("%sT%", st)
 290    furl = furl.replace("%st%", st.lower())
 291    furl = furl.replace(" ", "_")
 292    forecast = get_url(furl)
 293    if verbose: return forecast
 294    else:
 295       lines = forecast.split("\n")
 296       output = []
 297       if not quiet: output += lines[2:4]
 298       flines = int(flines)
 299       if not flines: flines = len(lines) - 5
 300       for line in lines[5:flines+5]:
 301          if imperial: line = filter_units(line, units="imperial")
 302          elif metric: line = filter_units(line, units="metric")
 303          if line.startswith("."):
 304             if quiet: output.append(line.replace(".", "", 1))
 305             else: output.append(line.replace(".", "   ", 1))
 306       return "\n".join(output)
 307
 308 def get_options(config):
 309    """Parse the options passed on the command line."""
 310
 311    # for optparse's builtin -h/--help option
 312    usage = "usage: %prog [ options ] [ alias [ alias [...] ] ]"
 313
 314    # for optparse's builtin --version option
 315    verstring = "%prog " + version
 316
 317    # create the parser
 318    import optparse
 319    option_parser = optparse.OptionParser(usage=usage, version=verstring)
 320
 321    # the -a/--alert option
 322    if config.has_option("default", "alert"):
 323       default_alert = bool(config.get("default", "alert"))
 324    else: default_alert = False
 325    option_parser.add_option("-a", "--alert",
 326       dest="alert",
 327       action="store_true",
 328       default=default_alert,
 329       help="include local alert notices")
 330
 331    # the --atypes option
 332    if config.has_option("default", "atypes"):
 333       default_atypes = config.get("default", "atypes")
 334    else:
 335       default_atypes = \
 336          "flash_flood/statement," \
 337          + "flash_flood/warning," \
 338          + "flash_flood/watch," \
 339          + "flood/coastal," \
 340          + "flood/statement," \
 341          + "flood/warning," \
 342          + "non_precip," \
 343          + "severe_weather_stmt," \
 344          + "special_weather_stmt," \
 345          + "thunderstorm," \
 346          + "tornado," \
 347          + "urgent_weather_message"
 348    option_parser.add_option("--atypes",
 349       dest="atypes",
 350       default=default_atypes,
 351       help="alert notification types to display")
 352
 353    # the --aurl option
 354    if config.has_option("default", "aurl"):
 355       default_aurl = config.get("default", "aurl")
 356    else:
 357       default_aurl = \
 358          "http://weather.noaa.gov/pub/data/watches_warnings/%atype%/%zone%.txt"
 359    option_parser.add_option("--aurl",
 360       dest="aurl",
 361       default=default_aurl,
 362       help="alert URL (including %atype% and %zone%)")
 363
 364    # separate options object from list of arguments and return both
 365    # the -c/--city option
 366    if config.has_option("default", "city"):
 367       default_city = config.get("default", "city")
 368    else: default_city = ""
 369    option_parser.add_option("-c", "--city",
 370       dest="city",
 371       default=default_city,
 372       help="the city name (ex: \"Raleigh Durham\")")
 373
 374    # the --flines option
 375    if config.has_option("default", "flines"):
 376       default_flines = config.get("default", "flines")
 377    else: default_flines = "0"
 378    option_parser.add_option("--flines",
 379       dest="flines",
 380       default=default_flines,
 381       help="maximum number of forecast lines to show")
 382
 383    # the -f/--forecast option
 384    if config.has_option("default", "forecast"):
 385       default_forecast = bool(config.get("default", "forecast"))
 386    else: default_forecast = False
 387    option_parser.add_option("-f", "--forecast",
 388       dest="forecast",
 389       action="store_true",
 390       default=default_forecast,
 391       help="include a local forecast")
 392
 393    # the --furl option
 394    if config.has_option("default", "furl"):
 395       default_furl = config.get("default", "furl")
 396    else:
 397       default_furl = \
 398          "http://weather.noaa.gov/pub/data/forecasts/city/%st%/%city%.txt"
 399    option_parser.add_option("--furl",
 400       dest="furl",
 401       default=default_furl,
 402       help="forecast URL (including %city% and %st%)")
 403
 404    # the --headers option
 405    if config.has_option("default", "headers"):
 406       default_headers = config.get("default", "headers")
 407    else:
 408       default_headers = \
 409          "temperature," \
 410          + "relative_humidity," \
 411          + "wind," \
 412          + "weather," \
 413          + "sky_conditions," \
 414          + "precipitation_last_hour"
 415    option_parser.add_option("--headers",
 416       dest="headers",
 417       default=default_headers,
 418       help="the conditions headers to display")
 419
 420    # the -i/--id option
 421    if config.has_option("default", "id"):
 422       default_id = config.get("default", "id")
 423    else: default_id = ""
 424    option_parser.add_option("-i", "--id",
 425       dest="id",
 426       default=default_id,
 427       help="the METAR station ID (ex: KRDU)")
 428
 429    # the --imperial option
 430    if config.has_option("default", "imperial"):
 431       default_imperial = bool(config.get("default", "imperial"))
 432    else: default_imperial = False
 433    option_parser.add_option("--imperial",
 434       dest="imperial",
 435       action="store_true",
 436       default=default_imperial,
 437       help="filter/convert for US/UK units")
 438
 439    # the -l/--list option
 440    option_parser.add_option("-l", "--list",
 441       dest="list",
 442       action="store_true",
 443       default=False,
 444       help="print a list of configured aliases")
 445
 446    # the -m/--metric option
 447    if config.has_option("default", "metric"):
 448       default_metric = bool(config.get("default", "metric"))
 449    else: default_metric = False
 450    option_parser.add_option("-m", "--metric",
 451       dest="metric",
 452       action="store_true",
 453       default=default_metric,
 454       help="filter/convert for metric units")
 455
 456    # the --murl option
 457    if config.has_option("default", "murl"):
 458       default_murl = config.get("default", "murl")
 459    else:
 460       default_murl = \
 461          "http://weather.noaa.gov/pub/data/observations/metar/decoded/%ID%.TXT"
 462    option_parser.add_option("--murl",
 463       dest="murl",
 464       default=default_murl,
 465       help="METAR URL (including %id%)")
 466
 467    # the -n/--no-conditions option
 468    if config.has_option("default", "conditions"):
 469       default_conditions = bool(config.get("default", "conditions"))
 470    else: default_conditions = True
 471    option_parser.add_option("-n", "--no-conditions",
 472       dest="conditions",
 473       action="store_false",
 474       default=default_conditions,
 475       help="disable output of current conditions (forces -f)")
 476
 477    # the -o/--omit-forecast option
 478    option_parser.add_option("-o", "--omit-forecast",
 479       dest="forecast",
 480       action="store_false",
 481       default=default_forecast,
 482       help="omit the local forecast (cancels -f)")
 483
 484    # the -q/--quiet option
 485    if config.has_option("default", "quiet"):
 486       default_quiet = bool(config.get("default", "quiet"))
 487    else: default_quiet = False
 488    option_parser.add_option("-q", "--quiet",
 489       dest="quiet",
 490       action="store_true",
 491       default=default_quiet,
 492       help="skip preambles and don't indent")
 493
 494    # the -s/--st option
 495    if config.has_option("default", "st"):
 496       default_st = config.get("default", "st")
 497    else: default_st = ""
 498    option_parser.add_option("-s", "--st",
 499       dest="st",
 500       default=default_st,
 501       help="the state abbreviation (ex: NC)")
 502
 503    # the -v/--verbose option
 504    if config.has_option("default", "verbose"):
 505       default_verbose = bool(config.get("default", "verbose"))
 506    else: default_verbose = False
 507    option_parser.add_option("-v", "--verbose",
 508       dest="verbose",
 509       action="store_true",
 510       default=default_verbose,
 511       help="show full decoded feeds (cancels -q)")
 512
 513    # the -z/--zones option
 514    if config.has_option("default", "zones"):
 515       default_zones = config.get("default", "zones")
 516    else: default_zones = ""
 517    option_parser.add_option("-z", "--zones",
 518       dest="zones",
 519       default=default_zones,
 520       help="alert zones (ex: nc/ncc183,nc/ncz041)")
 521
 522    options, arguments = option_parser.parse_args()
 523    return options, arguments
 524
 525 def get_config():
 526    """Parse the aliases and configuration."""
 527    import ConfigParser
 528    config = ConfigParser.ConfigParser()
 529    import os.path
 530    rcfiles = [
 531       "/etc/weatherrc",
 532       os.path.expanduser("~/.weatherrc"),
 533       "weatherrc"
 534       ]
 535    import os
 536    for rcfile in rcfiles:
 537       if os.access(rcfile, os.R_OK): config.read(rcfile)
 538    for section in config.sections():
 539       if section != section.lower():
 540          if config.has_section(section.lower()):
 541             config.remove_section(section.lower())
 542          config.add_section(section.lower())
 543          for option,value in config.items(section):
 544             config.set(section.lower(), option, value)
 545    return config
 546
 547 def list_aliases(config):
 548    """Return a formatted list of aliases defined in the config."""
 549    sections = []
 550    for section in sorted(config.sections()):
 551       if section.lower() not in sections and section != "default":
 552          sections.append(section.lower())
 553    output = "configured aliases..."
 554    for section in sorted(sections):
 555       output += "\n   " \
 556          + section \
 557          + ": --id=" \
 558          + quote(config.get(section, "id")) \
 559          + " --city=" \
 560          + quote(config.get(section, "city")) \
 561          + " --st=" \
 562          + quote(config.get(section, "st"))
 563    return output
 564