"""Contains various object definitions needed by the weather utility.""" weather_copyright = """\ # Copyright (c) 2006-2024 Jeremy Stanley . Permission to # use, copy, modify, and distribute this software is granted under terms # provided in the LICENSE file distributed with this software. #""" weather_version = "2.5.0" radian_to_km = 6372.795484 radian_to_mi = 3959.871528 class Selections: """An object to contain selection data.""" def __init__(self): """Store the config, options and arguments.""" self.config = get_config() self.options, self.arguments = get_options(self.config) if self.get_bool("cache") and self.get_bool("cache_search") \ and not self.get_bool("longlist"): integrate_search_cache( self.config, self.get("cachedir"), self.get("setpath") ) if not self.arguments: if "id" in self.options.__dict__ \ and self.options.__dict__["id"]: self.arguments.append( self.options.__dict__["id"] ) del( self.options.__dict__["id"] ) import sys message = "WARNING: the --id option is deprecated and will eventually be removed\n" sys.stderr.write(message) elif "city" in self.options.__dict__ \ and self.options.__dict__["city"] \ and "st" in self.options.__dict__ \ and self.options.__dict__["st"]: self.arguments.append( "^%s city, %s" % ( self.options.__dict__["city"], self.options.__dict__["st"] ) ) del( self.options.__dict__["city"] ) del( self.options.__dict__["st"] ) import sys message = "WARNING: the --city/--st options are deprecated and will eventually be removed\n" sys.stderr.write(message) def get(self, option, argument=None): """Retrieve data from the config or options.""" if argument: if self.config.has_section(argument) and ( self.config.has_option(argument, "city") \ or self.config.has_option(argument, "id") \ or self.config.has_option(argument, "st") ): self.config.remove_section(argument) import sys message = "WARNING: the city/id/st options are now unsupported in aliases\n" sys.stderr.write(message) if not self.config.has_section(argument): guessed = guess( argument, path=self.get("setpath"), info=self.get("info"), cache_search=( self.get("cache") and self.get("cache_search") ), cachedir=self.get("cachedir"), quiet=self.get_bool("quiet") ) self.config.add_section(argument) for item in guessed.items(): self.config.set(argument, *item) if self.config.has_option(argument, option): return self.config.get(argument, option) if option in self.options.__dict__: return self.options.__dict__[option] import sys message = "WARNING: no URI defined for %s\n" % option sys.stderr.write(message) return None def get_bool(self, option, argument=None): """Get data and coerce to a boolean if necessary.""" # Mimic configparser's getboolean() method by treating # false/no/off/0 as False and true/yes/on/1 as True values, # case-insensitively value = self.get(option, argument) if isinstance(value, bool): return value if isinstance(value, str): vlower = value.lower() if vlower in ('false', 'no', 'off', '0'): return False elif vlower in ('true', 'yes', 'on', '1'): return True raise ValueError("Not a boolean: %s" % value) def getint(self, option, argument=None): """Get data and coerce to an integer if necessary.""" value = self.get(option, argument) if value: return int(value) else: return 0 def average(coords): """Average a list of coordinates.""" x = 0 y = 0 for coord in coords: x += coord[0] y += coord[1] count = len(coords) return (x/count, y/count) def filter_units(line, units="imperial"): """Filter or convert units in a line of text between US/UK and metric.""" import re # filter lines with both pressures in the form of "X inches (Y hPa)" or # "X in. Hg (Y hPa)" dual_p = re.match( r"(.* )(\d*(\.\d+)? (inches|in\. Hg)) \((\d*(\.\d+)? hPa)\)(.*)", line ) if dual_p: preamble, in_hg, i_fr, i_un, hpa, h_fr, trailer = dual_p.groups() if units == "imperial": line = preamble + in_hg + trailer elif units == "metric": line = preamble + hpa + trailer # filter lines with both temperatures in the form of "X F (Y C)" dual_t = re.match( r"(.* )(-?\d*(\.\d+)? F) \((-?\d*(\.\d+)? C)\)(.*)", line ) if dual_t: preamble, fahrenheit, f_fr, celsius, c_fr, trailer = dual_t.groups() if units == "imperial": line = preamble + fahrenheit + trailer elif units == "metric": line = preamble + celsius + trailer # if metric is desired, convert distances in the form of "X mile(s)" to # "Y kilometer(s)" if units == "metric": imperial_d = re.match( r"(.* )(\d+)( mile\(s\))(.*)", line ) if imperial_d: preamble, mi, m_u, trailer = imperial_d.groups() line = preamble + str(int(round(int(mi)*1.609344))) \ + " kilometer(s)" + trailer # filter speeds in the form of "X MPH (Y KT)" to just "X MPH"; if metric is # desired, convert to "Z KPH" imperial_s = re.match( r"(.* )(\d+)( MPH)( \(\d+ KT\))(.*)", line ) if imperial_s: preamble, mph, m_u, kt, trailer = imperial_s.groups() if units == "imperial": line = preamble + mph + m_u + trailer elif units == "metric": line = preamble + str(int(round(int(mph)*1.609344))) + " KPH" + \ trailer imperial_s = re.match( r"(.* )(\d+)( MPH)( \(\d+ KT\))(.*)", line ) if imperial_s: preamble, mph, m_u, kt, trailer = imperial_s.groups() if units == "imperial": line = preamble + mph + m_u + trailer elif units == "metric": line = preamble + str(int(round(int(mph)*1.609344))) + " KPH" + \ trailer # if imperial is desired, qualify given forcast temperatures like "X F"; if # metric is desired, convert to "Y C" imperial_t = re.match( r"(.* )(High |high |Low |low )(\d+)(\.|,)(.*)", line ) if imperial_t: preamble, parameter, fahrenheit, sep, trailer = imperial_t.groups() if units == "imperial": line = preamble + parameter + fahrenheit + " F" + sep + trailer elif units == "metric": line = preamble + parameter \ + str(int(round((int(fahrenheit)-32)*5/9))) + " C" + sep \ + trailer # hand off the resulting line return line def get_uri( uri, ignore_fail=False, cache_data=False, cacheage=900, cachedir="." ): """Return a string containing the results of a URI GET.""" import os, time, urllib, urllib.error, urllib.request if cache_data: dcachedir = os.path.join( os.path.expanduser(cachedir), "datacache" ) if not os.path.exists(dcachedir): try: os.makedirs(dcachedir) except (IOError, OSError): pass dcache_fn = os.path.join( dcachedir, uri.split(":",1)[1].replace("/","_") ) now = time.time() if cache_data and os.access(dcache_fn, os.R_OK) \ and now-cacheage < os.stat(dcache_fn).st_mtime <= now: dcache_fd = open(dcache_fn) data = dcache_fd.read() dcache_fd.close() else: try: data = urllib.request.urlopen(uri).read().decode("utf-8") except urllib.error.URLError: if ignore_fail: return "" import os, sys sys.stderr.write("%s error: failed to retrieve\n %s\n\n" % ( os.path.basename( sys.argv[0] ), uri)) raise # Some data sources are HTML with the plain text wrapped in pre tags if "
" in data:
            data = data[data.find("
")+5:data.find("
")] if cache_data: try: import codecs dcache_fd = codecs.open(dcache_fn, "w", "utf-8") dcache_fd.write(data) dcache_fd.close() except (IOError, OSError): pass return data def get_metar( uri=None, verbose=False, quiet=False, headers=None, imperial=False, metric=False, cache_data=False, cacheage=900, cachedir="." ): """Return a summarized METAR for the specified station.""" if not uri: import os, sys message = "%s error: METAR URI required for conditions\n" % \ os.path.basename( sys.argv[0] ) sys.stderr.write(message) sys.exit(1) metar = get_uri( uri, cache_data=cache_data, cacheage=cacheage, cachedir=cachedir ) if type(metar) is bytes: metar = metar.decode("utf-8") if verbose: return metar else: import re lines = metar.split("\n") if not headers: headers = \ "relative_humidity," \ + "precipitation_last_hour," \ + "sky conditions," \ + "temperature," \ + "heat index," \ + "windchill," \ + "weather," \ + "wind" headerlist = headers.lower().replace("_"," ").split(",") output = [] if not quiet: title = "Current conditions at %s" place = lines[0].split(", ") if len(place) > 1: place = "%s, %s" % ( place[0].title(), place[1] ) else: place = "" output.append(title%place) output.append("Last updated " + lines[1]) header_match = False for header in headerlist: for line in lines: if line.lower().startswith(header + ":"): if re.match(r".*:\d+$", line): line = line[:line.rfind(":")] if imperial: line = filter_units(line, units="imperial") elif metric: line = filter_units(line, units="metric") if quiet: output.append(line) else: output.append(" " + line) header_match = True if not header_match: output.append( "(no conditions matched your header list, try with --verbose)" ) return "\n".join(output) def get_alert( uri=None, verbose=False, quiet=False, cache_data=False, cacheage=900, cachedir=".", delay=1 ): """Return alert notice for the specified URI.""" if not uri: return "" alert = get_uri( uri, ignore_fail=True, cache_data=cache_data, cacheage=cacheage, cachedir=cachedir ).strip() if type(alert) is bytes: alert = alert.decode("utf-8") if alert: if verbose: return alert else: import re if re.search(r"\nNational Weather Service", alert): muted = True else: muted = False expirycheck = re.search(r"Expires:([0-9]{12})", alert) if expirycheck: # only report alerts and forecasts that expired less than delay # hours ago import datetime, zoneinfo expiration = datetime.datetime.fromisoformat( "%s-%s-%sT%s:%s" % ( expirycheck[1][:4], expirycheck[1][4:6], expirycheck[1][6:8], expirycheck[1][8:10], expirycheck[1][-2:], )).replace(tzinfo=zoneinfo.ZoneInfo("UTC")) now = datetime.datetime.now(tz=zoneinfo.ZoneInfo("UTC")) if now - expiration > datetime.timedelta(hours=delay): return "" lines = alert.split("\n") output = [] for line in lines: if muted and line.startswith("National Weather Service"): muted = False line = "" elif line == "&&": line = "" elif line == "$$": muted = True if line and not muted: if quiet: output.append(line) else: output.append(" " + line) return "\n".join(output) def get_options(config): """Parse the options passed on the command line.""" # for optparse's builtin -h/--help option usage = \ "usage: %prog [options] [alias1|search1 [alias2|search2 [...]]]" # for optparse's builtin --version option verstring = "%prog " + weather_version # create the parser import optparse option_parser = optparse.OptionParser(usage=usage, version=verstring) # separate options object from list of arguments and return both # the -a/--alert option if config.has_option("default", "alert"): default_alert = config.getboolean("default", "alert") else: default_alert = False option_parser.add_option("-a", "--alert", dest="alert", action="store_true", default=default_alert, help="include local alert notices") # the --atypes option if config.has_option("default", "atypes"): default_atypes = config.get("default", "atypes") else: default_atypes = \ "coastal_flood_statement," \ + "flash_flood_statement," \ + "flash_flood_warning," \ + "flash_flood_watch," \ + "flood_warning," \ + "severe_thunderstorm_warning," \ + "severe_weather_statement," \ + "special_weather_statement," \ + "tornado," \ + "urgent_weather_message" option_parser.add_option("--atypes", dest="atypes", default=default_atypes, help="list of alert notification types to display") # the --build-sets option option_parser.add_option("--build-sets", dest="build_sets", action="store_true", default=False, help="(re)build location correlation sets") # the --cacheage option if config.has_option("default", "cacheage"): default_cacheage = config.getint("default", "cacheage") else: default_cacheage = 900 option_parser.add_option("--cacheage", dest="cacheage", default=default_cacheage, help="duration in seconds to refresh cached data") # the --cachedir option if config.has_option("default", "cachedir"): default_cachedir = config.get("default", "cachedir") else: default_cachedir = "~/.weather" option_parser.add_option("--cachedir", dest="cachedir", default=default_cachedir, help="directory for storing cached searches and data") # the --delay option if config.has_option("default", "delay"): default_delay = config.getint("default", "delay") else: default_delay = 1 option_parser.add_option("--delay", dest="delay", default=default_delay, help="hours to delay alert and forecast expiration") # the -f/--forecast option if config.has_option("default", "forecast"): default_forecast = config.getboolean("default", "forecast") else: default_forecast = False option_parser.add_option("-f", "--forecast", dest="forecast", action="store_true", default=default_forecast, help="include a local forecast") # the --headers option if config.has_option("default", "headers"): default_headers = config.get("default", "headers") else: default_headers = \ "temperature," \ + "relative_humidity," \ + "wind," \ + "heat_index," \ + "windchill," \ + "weather," \ + "sky_conditions," \ + "precipitation_last_hour" option_parser.add_option("--headers", dest="headers", default=default_headers, help="list of conditions headers to display") # the --imperial option if config.has_option("default", "imperial"): default_imperial = config.getboolean("default", "imperial") else: default_imperial = False option_parser.add_option("--imperial", dest="imperial", action="store_true", default=default_imperial, help="filter/convert conditions for US/UK units") # the --info option option_parser.add_option("--info", dest="info", action="store_true", default=False, help="output detailed information for your search") # the -l/--list option option_parser.add_option("-l", "--list", dest="list", action="store_true", default=False, help="list all configured aliases and cached searches") # the --longlist option option_parser.add_option("--longlist", dest="longlist", action="store_true", default=False, help="display details of all configured aliases") # the -m/--metric option if config.has_option("default", "metric"): default_metric = config.getboolean("default", "metric") else: default_metric = False option_parser.add_option("-m", "--metric", dest="metric", action="store_true", default=default_metric, help="filter/convert conditions for metric units") # the -n/--no-conditions option if config.has_option("default", "conditions"): default_conditions = config.getboolean("default", "conditions") else: default_conditions = True option_parser.add_option("-n", "--no-conditions", dest="conditions", action="store_false", default=default_conditions, help="disable output of current conditions") # the --no-cache option if config.has_option("default", "cache"): default_cache = config.getboolean("default", "cache") else: default_cache = True option_parser.add_option("--no-cache", dest="cache", action="store_false", default=True, help="disable all caching (searches and data)") # the --no-cache-data option if config.has_option("default", "cache_data"): default_cache_data = config.getboolean("default", "cache_data") else: default_cache_data = True option_parser.add_option("--no-cache-data", dest="cache_data", action="store_false", default=True, help="disable retrieved data caching") # the --no-cache-search option if config.has_option("default", "cache_search"): default_cache_search = config.getboolean("default", "cache_search") else: default_cache_search = True option_parser.add_option("--no-cache-search", dest="cache_search", action="store_false", default=True, help="disable search result caching") # the -q/--quiet option if config.has_option("default", "quiet"): default_quiet = config.getboolean("default", "quiet") else: default_quiet = False option_parser.add_option("-q", "--quiet", dest="quiet", action="store_true", default=default_quiet, help="skip preambles and don't indent") # the --setpath option if config.has_option("default", "setpath"): default_setpath = config.get("default", "setpath") else: default_setpath = ".:~/.weather" option_parser.add_option("--setpath", dest="setpath", default=default_setpath, help="directory search path for correlation sets") # the -v/--verbose option if config.has_option("default", "verbose"): default_verbose = config.getboolean("default", "verbose") else: default_verbose = False option_parser.add_option("-v", "--verbose", dest="verbose", action="store_true", default=default_verbose, help="show full decoded feeds") # deprecated options if config.has_option("default", "city"): default_city = config.get("default", "city") else: default_city = "" option_parser.add_option("-c", "--city", dest="city", default=default_city, help=optparse.SUPPRESS_HELP) if config.has_option("default", "id"): default_id = config.get("default", "id") else: default_id = "" option_parser.add_option("-i", "--id", dest="id", default=default_id, help=optparse.SUPPRESS_HELP) if config.has_option("default", "st"): default_st = config.get("default", "st") else: default_st = "" option_parser.add_option("-s", "--st", dest="st", default=default_st, help=optparse.SUPPRESS_HELP) options, arguments = option_parser.parse_args() return options, arguments def get_config(): """Parse the aliases and configuration.""" import configparser, os config = configparser.ConfigParser() rcfiles = [ "/etc/weatherrc", "/etc/weather/weatherrc", os.path.expanduser("~/.weather/weatherrc"), os.path.expanduser("~/.weatherrc"), "weatherrc" ] for rcfile in rcfiles: if os.access(rcfile, os.R_OK): config.read(rcfile, encoding="utf-8") for section in config.sections(): if section != section.lower(): if config.has_section(section.lower()): config.remove_section(section.lower()) config.add_section(section.lower()) for option,value in config.items(section): config.set(section.lower(), option, value) return config def integrate_search_cache(config, cachedir, setpath): """Add cached search results into the configuration.""" import configparser, os, time scache_fn = os.path.join( os.path.expanduser(cachedir), "searches" ) if not os.access(scache_fn, os.R_OK): return config scache_fd = open(scache_fn) created = float( scache_fd.readline().split(":")[1].strip().split()[0] ) scache_fd.close() now = time.time() datafiles = data_index(setpath) if datafiles: data_freshness = sorted( [ x[1] for x in datafiles.values() ], reverse=True )[0] else: data_freshness = now if created < data_freshness <= now: try: os.remove(scache_fn) print( "[clearing outdated %s]" % scache_fn ) except (IOError, OSError): pass return config scache = configparser.ConfigParser() scache.read(scache_fn, encoding="utf-8") for section in scache.sections(): if not config.has_section(section): config.add_section(section) for option,value in scache.items(section): config.set(section, option, value) return config def list_aliases(config, detail=False): """Return a formatted list of aliases defined in the config.""" if detail: output = "\n# configured alias details..." for section in sorted(config.sections()): output += "\n\n[%s]" % section for item in sorted(config.items(section)): output += "\n%s = %s" % item output += "\n" else: output = "configured aliases and cached searches..." for section in sorted(config.sections()): if config.has_option(section, "description"): description = config.get(section, "description") else: description = "(no description provided)" output += "\n %s: %s" % (section, description) return output def data_index(path): import os datafiles = {} for filename in ("airports", "places", "stations", "zctas", "zones"): for dirname in path.split(":"): for extension in ("", ".gz", ".txt"): candidate = os.path.expanduser( os.path.join( dirname, "".join( (filename, extension) ) ) ) if os.path.exists(candidate): datafiles[filename] = ( candidate, os.stat(candidate).st_mtime ) break if filename in datafiles: break return datafiles def guess( expression, path=".", max_results=20, info=False, cache_search=False, cacheage=900, cachedir=".", quiet=False ): """Find URIs using airport, gecos, placename, station, ZCTA/ZIP, zone.""" import codecs, configparser, datetime, time, os, re, sys datafiles = data_index(path) if re.match("[A-Za-z]{3}$", expression): searchtype = "airport" elif re.match("[A-Za-z0-9]{4}$", expression): searchtype = "station" elif re.match("[A-Za-z]{2}[Zz][0-9]{3}$", expression): searchtype = "zone" elif re.match("[0-9]{5}$", expression): searchtype = "ZCTA" elif re.match( r"[\+-]?\d+(\.\d+)?(-\d+){,2}[ENSWensw]?, *[\+-]?\d+(\.\d+)?(-\d+){,2}[ENSWensw]?$", expression ): searchtype = "coordinates" elif re.match(r"(FIPS|fips)\d+$", expression): searchtype = "FIPS" else: searchtype = "name" cache_search = False if cache_search: action = "caching" else: action = "using" if info: scores = [ (0.005, "bad"), (0.025, "poor"), (0.160, "suspect"), (0.500, "mediocre"), (0.840, "good"), (0.975, "great"), (0.995, "excellent"), (1.000, "ideal"), ] if not quiet: print("Searching via %s..."%searchtype) stations = configparser.ConfigParser() dataname = "stations" if dataname in datafiles: datafile = datafiles[dataname][0] if datafile.endswith(".gz"): import gzip stations.read_string( gzip.open(datafile).read().decode("utf-8") ) else: stations.read(datafile, encoding="utf-8") else: message = "%s error: can't find \"%s\" data file\n" % ( os.path.basename( sys.argv[0] ), dataname ) sys.stderr.write(message) exit(1) zones = configparser.ConfigParser() dataname = "zones" if dataname in datafiles: datafile = datafiles[dataname][0] if datafile.endswith(".gz"): import gzip zones.read_string( gzip.open(datafile).read().decode("utf-8") ) else: zones.read(datafile, encoding="utf-8") else: message = "%s error: can't find \"%s\" data file\n" % ( os.path.basename( sys.argv[0] ), dataname ) sys.stderr.write(message) exit(1) search = None station = ("", 0) zone = ("", 0) dataset = None possibilities = [] uris = {} if searchtype == "airport": expression = expression.lower() airports = configparser.ConfigParser() dataname = "airports" if dataname in datafiles: datafile = datafiles[dataname][0] if datafile.endswith(".gz"): import gzip airports.read_string( gzip.open(datafile).read().decode("utf-8") ) else: airports.read(datafile, encoding="utf-8") else: message = "%s error: can't find \"%s\" data file\n" % ( os.path.basename( sys.argv[0] ), dataname ) sys.stderr.write(message) exit(1) if airports.has_section(expression) \ and airports.has_option(expression, "station"): search = (expression, "IATA/FAA airport code %s" % expression) station = ( airports.get(expression, "station"), 0 ) if stations.has_option(station[0], "zone"): zone = eval( stations.get(station[0], "zone") ) dataset = stations if not ( info or quiet ) \ and stations.has_option( station[0], "description" ): print( "[%s result %s]" % ( action, stations.get(station[0], "description") ) ) else: message = "No IATA/FAA airport code \"%s\" in the %s file.\n" % ( expression, datafiles["airports"][0] ) sys.stderr.write(message) exit(1) elif searchtype == "station": expression = expression.lower() if stations.has_section(expression): station = (expression, 0) if not search: search = (expression, "ICAO station code %s" % expression) if stations.has_option(expression, "zone"): zone = eval( stations.get(expression, "zone") ) dataset = stations if not ( info or quiet ) \ and stations.has_option(expression, "description"): print( "[%s result %s]" % ( action, stations.get(expression, "description") ) ) else: message = "No ICAO weather station \"%s\" in the %s file.\n" % ( expression, datafiles["stations"][0] ) sys.stderr.write(message) exit(1) elif searchtype == "zone": expression = expression.lower() if zones.has_section(expression) \ and zones.has_option(expression, "station"): zone = (expression, 0) station = eval( zones.get(expression, "station") ) dataset = zones search = (expression, "NWS/NOAA weather zone %s" % expression) if not ( info or quiet ) \ and zones.has_option(expression, "description"): print( "[%s result %s]" % ( action, zones.get(expression, "description") ) ) else: message = "No usable NWS weather zone \"%s\" in the %s file.\n" % ( expression, datafiles["zones"][0] ) sys.stderr.write(message) exit(1) elif searchtype == "ZCTA": zctas = configparser.ConfigParser() dataname = "zctas" if dataname in datafiles: datafile = datafiles[dataname][0] if datafile.endswith(".gz"): import gzip zctas.read_string( gzip.open(datafile).read().decode("utf-8") ) else: zctas.read(datafile, encoding="utf-8") else: message = "%s error: can't find \"%s\" data file\n" % ( os.path.basename( sys.argv[0] ), dataname ) sys.stderr.write(message) exit(1) dataset = zctas if zctas.has_section(expression) \ and zctas.has_option(expression, "station"): station = eval( zctas.get(expression, "station") ) search = (expression, "Census ZCTA (ZIP code) %s" % expression) if zctas.has_option(expression, "zone"): zone = eval( zctas.get(expression, "zone") ) else: message = "No census ZCTA (ZIP code) \"%s\" in the %s file.\n" % ( expression, datafiles["zctas"][0] ) sys.stderr.write(message) exit(1) elif searchtype == "coordinates": search = (expression, "Geographic coordinates %s" % expression) stationtable = {} for station in stations.sections(): if stations.has_option(station, "location"): stationtable[station] = { "location": eval( stations.get(station, "location") ) } station = closest( gecos(expression), stationtable, "location", 0.1 ) if not station[0]: message = "No ICAO weather station found near %s.\n" % expression sys.stderr.write(message) exit(1) zonetable = {} for zone in zones.sections(): if zones.has_option(zone, "centroid"): zonetable[zone] = { "centroid": eval( zones.get(zone, "centroid") ) } zone = closest( gecos(expression), zonetable, "centroid", 0.1 ) if not zone[0]: message = "No NWS weather zone near %s; forecasts unavailable.\n" \ % expression sys.stderr.write(message) elif searchtype in ("FIPS", "name"): places = configparser.ConfigParser() dataname = "places" if dataname in datafiles: datafile = datafiles[dataname][0] if datafile.endswith(".gz"): import gzip places.read_string( gzip.open(datafile).read().decode("utf-8") ) else: places.read(datafile, encoding="utf-8") else: message = "%s error: can't find \"%s\" data file\n" % ( os.path.basename( sys.argv[0] ), dataname ) sys.stderr.write(message) exit(1) dataset = places place = expression.lower() if places.has_section(place) and places.has_option(place, "station"): station = eval( places.get(place, "station") ) search = (expression, "Census Place %s" % expression) if places.has_option(place, "description"): search = ( search[0], search[1] + ", %s" % places.get(place, "description") ) if places.has_option(place, "zone"): zone = eval( places.get(place, "zone") ) if not ( info or quiet ) \ and places.has_option(place, "description"): print( "[%s result %s]" % ( action, places.get(place, "description") ) ) else: for place in places.sections(): if places.has_option(place, "description") \ and places.has_option(place, "station") \ and re.search( expression, places.get(place, "description"), re.I ): possibilities.append(place) for place in stations.sections(): if stations.has_option(place, "description") \ and re.search( expression, stations.get(place, "description"), re.I ): possibilities.append(place) for place in zones.sections(): if zones.has_option(place, "description") \ and zones.has_option(place, "station") \ and re.search( expression, zones.get(place, "description"), re.I ): possibilities.append(place) if len(possibilities) == 1: place = possibilities[0] if places.has_section(place): station = eval( places.get(place, "station") ) description = places.get(place, "description") if places.has_option(place, "zone"): zone = eval( places.get(place, "zone" ) ) search = ( expression, "%s: %s" % (place, description) ) elif stations.has_section(place): station = (place, 0.0) description = stations.get(place, "description") if stations.has_option(place, "zone"): zone = eval( stations.get(place, "zone" ) ) search = ( expression, "ICAO station code %s" % place ) elif zones.has_section(place): station = eval( zones.get(place, "station") ) description = zones.get(place, "description") zone = (place, 0.0) search = ( expression, "NWS/NOAA weather zone %s" % place ) if not ( info or quiet ): print( "[%s result %s]" % (action, description) ) if not possibilities and not station[0]: message = "No FIPS code/census area match in the %s file.\n" % ( datafiles["places"][0] ) sys.stderr.write(message) exit(1) if station[0]: uris["metar"] = stations.get( station[0], "metar" ) if zone[0]: for key,value in zones.items( zone[0] ): if key not in ("centroid", "description", "station"): uris[key] = value elif possibilities: count = len(possibilities) if count <= max_results: print( "Your search is ambiguous, returning %s matches:" % count ) for place in sorted(possibilities): if places.has_section(place): print( " [%s] %s" % ( place, places.get(place, "description") ) ) elif stations.has_section(place): print( " [%s] %s" % ( place, stations.get(place, "description") ) ) elif zones.has_section(place): print( " [%s] %s" % ( place, zones.get(place, "description") ) ) else: print( "Your search is too ambiguous, returning %s matches." % count ) exit(0) if info: stationlist = [] zonelist = [] if dataset: for section in dataset.sections(): if dataset.has_option(section, "station"): stationlist.append( eval( dataset.get(section, "station") )[1] ) if dataset.has_option(section, "zone"): zonelist.append( eval( dataset.get(section, "zone") )[1] ) stationlist.sort() zonelist.sort() scount = len(stationlist) zcount = len(zonelist) sranks = [] zranks = [] for score in scores: if stationlist: sranks.append( stationlist[ int( (1-score[0]) * scount ) ] ) if zonelist: zranks.append( zonelist[ int( (1-score[0]) * zcount ) ] ) description = search[1] uris["description"] = description print( "%s\n%s" % ( description, "-" * len(description) ) ) print( "%s: %s" % ( station[0], stations.get( station[0], "description" ) ) ) km = radian_to_km*station[1] mi = radian_to_mi*station[1] if sranks and not description.startswith("ICAO station code "): for index in range(0, len(scores)): if station[1] >= sranks[index]: score = scores[index][1] break print( " (proximity %s, %.3gkm, %.3gmi)" % ( score, km, mi ) ) elif searchtype == "coordinates": print( " (%.3gkm, %.3gmi)" % (km, mi) ) if zone[0]: print( "%s: %s" % ( zone[0], zones.get( zone[0], "description" ) ) ) km = radian_to_km*zone[1] mi = radian_to_mi*zone[1] if zranks and not description.startswith("NWS/NOAA weather zone "): for index in range(0, len(scores)): if zone[1] >= zranks[index]: score = scores[index][1] break print( " (proximity %s, %.3gkm, %.3gmi)" % ( score, km, mi ) ) elif searchtype == "coordinates" and zone[0]: print( " (%.3gkm, %.3gmi)" % (km, mi) ) if cache_search: now = time.time() nowstamp = "%s (%s)" % ( now, datetime.datetime.isoformat( datetime.datetime.fromtimestamp(now), " " ) ) search_cache = ["\n"] search_cache.append( "[%s]\n" % search[0] ) search_cache.append( "cached = %s\n" % nowstamp ) for uriname in sorted(uris.keys()): search_cache.append( "%s = %s\n" % ( uriname, uris[uriname] ) ) real_cachedir = os.path.expanduser(cachedir) if not os.path.exists(real_cachedir): try: os.makedirs(real_cachedir) except (IOError, OSError): pass scache_fn = os.path.join(real_cachedir, "searches") if not os.path.exists(scache_fn): then = sorted( [ x[1] for x in datafiles.values() ], reverse=True )[0] thenstamp = "%s (%s)" % ( then, datetime.datetime.isoformat( datetime.datetime.fromtimestamp(then), " " ) ) search_cache.insert( 0, "# based on data files from: %s\n" % thenstamp ) try: scache_existing = configparser.ConfigParser() scache_existing.read(scache_fn, encoding="utf-8") if not scache_existing.has_section(search[0]): scache_fd = codecs.open(scache_fn, "a", "utf-8") scache_fd.writelines(search_cache) scache_fd.close() except (IOError, OSError): pass if not info: return(uris) def closest(position, nodes, fieldname, angle=None): import math if not angle: angle = 2*math.pi match = None for name in nodes: if fieldname in nodes[name]: node = nodes[name][fieldname] if node and abs( position[0]-node[0] ) < angle: if abs( position[1]-node[1] ) < angle \ or abs( abs( position[1]-node[1] ) - 2*math.pi ) < angle: if position == node: angle = 0 match = name else: candidate = math.acos( math.sin( position[0] ) * math.sin( node[0] ) \ + math.cos( position[0] ) \ * math.cos( node[0] ) \ * math.cos( position[1] - node[1] ) ) if candidate < angle: angle = candidate match = name if match: match = str(match) return (match, angle) def gecos(formatted): import math, re coordinates = formatted.split(",") for coordinate in range(0, 2): degrees, foo, minutes, bar, seconds, hemisphere = re.match( r"([\+-]?\d+\.?\d*)(-(\d+))?(-(\d+))?([ensw]?)$", coordinates[coordinate].strip().lower() ).groups() value = float(degrees) if minutes: value += float(minutes)/60 if seconds: value += float(seconds)/3600 if hemisphere and hemisphere in "sw": value *= -1 coordinates[coordinate] = math.radians(value) return tuple(coordinates) def correlate(): import codecs, configparser, csv, datetime, hashlib, os, re, sys, time import zipfile, zoneinfo for filename in os.listdir("."): if re.match("[0-9]{4}_Gaz_counties_national.zip$", filename): gcounties_an = filename gcounties_fn = filename[:-4] + ".txt" elif re.match("[0-9]{4}_Gaz_cousubs_national.zip$", filename): gcousubs_an = filename gcousubs_fn = filename[:-4] + ".txt" elif re.match("[0-9]{4}_Gaz_place_national.zip$", filename): gplace_an = filename gplace_fn = filename[:-4] + ".txt" elif re.match("[0-9]{4}_Gaz_zcta_national.zip$", filename): gzcta_an = filename gzcta_fn = filename[:-4] + ".txt" elif re.match("bp[0-9]{2}[a-z]{2}[0-9]{2}.dbx$", filename): cpfzcf_fn = filename nsdcccc_fn = "nsd_cccc.txt" ourairports_fn = "airports.csv" overrides_fn = "overrides.conf" overrideslog_fn = "overrides.log" slist_fn = "slist" zlist_fn = "zlist" qalog_fn = "qa.log" airports_fn = "airports" places_fn = "places" stations_fn = "stations" zctas_fn = "zctas" zones_fn = "zones" header = """\ %s # generated by %s on %s from these public domain sources: # # https://www.census.gov/geographies/reference-files/time-series/geo/gazetteer-files.html # %s %s %s # %s %s %s # %s %s %s # %s %s %s # # https://www.weather.gov/gis/ZoneCounty/ # %s %s %s # # https://tgftp.nws.noaa.gov/data/ # %s %s %s # # https://ourairports.com/data/ # %s %s %s # # ...and these manually-generated or hand-compiled adjustments: # %s %s %s # %s %s %s # %s %s %s\ """ % ( weather_copyright, os.path.basename( sys.argv[0] ), datetime.date.isoformat( datetime.datetime.utcfromtimestamp( int(os.environ.get('SOURCE_DATE_EPOCH', time.time())) ) ), hashlib.md5( open(gcounties_an, "rb").read() ).hexdigest(), datetime.date.isoformat( datetime.datetime.utcfromtimestamp( os.path.getmtime(gcounties_an) ) ), gcounties_an, hashlib.md5( open(gcousubs_an, "rb").read() ).hexdigest(), datetime.date.isoformat( datetime.datetime.utcfromtimestamp( os.path.getmtime(gcousubs_an) ) ), gcousubs_an, hashlib.md5( open(gplace_an, "rb").read() ).hexdigest(), datetime.date.isoformat( datetime.datetime.utcfromtimestamp( os.path.getmtime(gplace_an) ) ), gplace_an, hashlib.md5( open(gzcta_an, "rb").read() ).hexdigest(), datetime.date.isoformat( datetime.datetime.utcfromtimestamp( os.path.getmtime(gzcta_an) ) ), gzcta_an, hashlib.md5( open(cpfzcf_fn, "rb").read() ).hexdigest(), datetime.date.isoformat( datetime.datetime.utcfromtimestamp( os.path.getmtime(cpfzcf_fn) ) ), cpfzcf_fn, hashlib.md5( open(nsdcccc_fn, "rb").read() ).hexdigest(), datetime.date.isoformat( datetime.datetime.utcfromtimestamp( os.path.getmtime(nsdcccc_fn) ) ), nsdcccc_fn, hashlib.md5( open(ourairports_fn, "rb").read() ).hexdigest(), datetime.date.isoformat( datetime.datetime.utcfromtimestamp( os.path.getmtime(ourairports_fn) ) ), ourairports_fn, hashlib.md5( open(overrides_fn, "rb").read() ).hexdigest(), datetime.date.isoformat( datetime.datetime.utcfromtimestamp( os.path.getmtime(overrides_fn) ) ), overrides_fn, hashlib.md5( open(slist_fn, "rb").read() ).hexdigest(), datetime.date.isoformat( datetime.datetime.utcfromtimestamp( os.path.getmtime(slist_fn) ) ), slist_fn, hashlib.md5( open(zlist_fn, "rb").read() ).hexdigest(), datetime.date.isoformat( datetime.datetime.utcfromtimestamp( os.path.getmtime(zlist_fn) ) ), zlist_fn ) airports = {} places = {} stations = {} zctas = {} zones = {} message = "Reading %s:%s..." % (gcounties_an, gcounties_fn) sys.stdout.write(message) sys.stdout.flush() count = 0 gcounties = zipfile.ZipFile(gcounties_an).open(gcounties_fn, "r") columns = gcounties.readline().decode("utf-8").strip().split("\t") for line in gcounties: fields = line.decode("utf-8").strip().split("\t") f_geoid = fields[ columns.index("GEOID") ].strip() f_name = fields[ columns.index("NAME") ].strip() f_usps = fields[ columns.index("USPS") ].strip() f_intptlat = fields[ columns.index("INTPTLAT") ].strip() f_intptlong = fields[ columns.index("INTPTLONG") ].strip() if f_geoid and f_name and f_usps and f_intptlat and f_intptlong: fips = "fips%s" % f_geoid if fips not in places: places[fips] = {} places[fips]["centroid"] = gecos( "%s,%s" % (f_intptlat, f_intptlong) ) places[fips]["description"] = "%s, %s" % (f_name, f_usps) count += 1 gcounties.close() print("done (%s lines)." % count) message = "Reading %s:%s..." % (gcousubs_an, gcousubs_fn) sys.stdout.write(message) sys.stdout.flush() count = 0 gcousubs = zipfile.ZipFile(gcousubs_an).open(gcousubs_fn, "r") columns = gcousubs.readline().decode("utf-8").strip().split("\t") for line in gcousubs: fields = line.decode("utf-8").strip().split("\t") f_geoid = fields[ columns.index("GEOID") ].strip() f_name = fields[ columns.index("NAME") ].strip() f_usps = fields[ columns.index("USPS") ].strip() f_intptlat = fields[ columns.index("INTPTLAT") ].strip() f_intptlong = fields[ columns.index("INTPTLONG") ].strip() if f_geoid and f_name and f_usps and f_intptlat and f_intptlong: fips = "fips%s" % f_geoid if fips not in places: places[fips] = {} places[fips]["centroid"] = gecos( "%s,%s" % (f_intptlat, f_intptlong) ) places[fips]["description"] = "%s, %s" % (f_name, f_usps) count += 1 gcousubs.close() print("done (%s lines)." % count) message = "Reading %s:%s..." % (gplace_an, gplace_fn) sys.stdout.write(message) sys.stdout.flush() count = 0 gplace = zipfile.ZipFile(gplace_an).open(gplace_fn, "r") columns = gplace.readline().decode("utf-8").strip().split("\t") for line in gplace: fields = line.decode("utf-8").strip().split("\t") f_geoid = fields[ columns.index("GEOID") ].strip() f_name = fields[ columns.index("NAME") ].strip() f_usps = fields[ columns.index("USPS") ].strip() f_intptlat = fields[ columns.index("INTPTLAT") ].strip() f_intptlong = fields[ columns.index("INTPTLONG") ].strip() if f_geoid and f_name and f_usps and f_intptlat and f_intptlong: fips = "fips%s" % f_geoid if fips not in places: places[fips] = {} places[fips]["centroid"] = gecos( "%s,%s" % (f_intptlat, f_intptlong) ) places[fips]["description"] = "%s, %s" % (f_name, f_usps) count += 1 gplace.close() print("done (%s lines)." % count) message = "Reading %s..." % slist_fn sys.stdout.write(message) sys.stdout.flush() count = 0 slist = codecs.open(slist_fn, "r", "utf-8") for line in slist: icao = line.split("#")[0].strip() if icao: stations[icao] = { "metar": "https://tgftp.nws.noaa.gov/data/observations/"\ + "metar/decoded/%s.TXT" % icao.upper() } count += 1 slist.close() print("done (%s lines)." % count) message = "Reading %s..." % nsdcccc_fn sys.stdout.write(message) sys.stdout.flush() count = 0 nsdcccc = codecs.open(nsdcccc_fn, "r", "utf-8") for line in nsdcccc: line = str(line) fields = line.split(";") icao = fields[0].strip().lower() if icao in stations: description = [] name = " ".join( fields[3].strip().title().split() ) if name: description.append(name) st = fields[4].strip() if st: description.append(st) country = " ".join( fields[5].strip().title().split() ) if country: description.append(country) if description: stations[icao]["description"] = ", ".join(description) lat, lon = fields[7:9] if lat and lon: stations[icao]["location"] = gecos( "%s,%s" % (lat, lon) ) elif "location" not in stations[icao]: lat, lon = fields[5:7] if lat and lon: stations[icao]["location"] = gecos( "%s,%s" % (lat, lon) ) count += 1 nsdcccc.close() print("done (%s lines)." % count) message = "Reading %s..." % ourairports_fn sys.stdout.write(message) sys.stdout.flush() count = 0 ourairports = open(ourairports_fn, "r") for row in csv.reader(ourairports): icao = row[12].lower() if icao in stations: iata = row[13].lower() if len(iata) == 3: airports[iata] = { "station": icao } if "description" not in stations[icao]: description = [] name = row[3] if name: description.append(name) municipality = row[10] if municipality: description.append(municipality) region = row[9] country = row[8] if region: if "-" in region: c,r = region.split("-", 1) if c == country: region = r description.append(region) if country: description.append(country) if description: stations[icao]["description"] = ", ".join(description) if "location" not in stations[icao]: lat = row[4] if lat: lon = row[5] if lon: stations[icao]["location"] = gecos( "%s,%s" % (lat, lon) ) count += 1 ourairports.close() print("done (%s lines)." % count) message = "Reading %s..." % zlist_fn sys.stdout.write(message) sys.stdout.flush() count = 0 zlist = codecs.open(zlist_fn, "r", "utf-8") for line in zlist: line = line.split("#")[0].strip() if line: zones[line] = {} count += 1 zlist.close() print("done (%s lines)." % count) message = "Reading %s..." % cpfzcf_fn sys.stdout.write(message) sys.stdout.flush() count = 0 cpfz = {} cpfzcf = codecs.open(cpfzcf_fn, "r", "utf-8") for line in cpfzcf: fields = line.strip().split("|") if len(fields) == 11 \ and fields[0] and fields[1] and fields[9] and fields[10]: zone = "z".join( fields[:2] ).lower() if zone in zones: state = fields[0] description = fields[3].strip() fips = "fips%s"%fields[6] countycode = "%sc%s" % (state.lower(), fips[-3:]) if state: zones[zone]["coastal_flood_statement"] = ( "https://tgftp.nws.noaa.gov/data/watches_warnings/" "flood/coastal/%s/%s.txt" % (state.lower(), zone)) zones[zone]["flash_flood_statement"] = ( "https://tgftp.nws.noaa.gov/data/watches_warnings/" "flash_flood/statement/%s/%s.txt" % (state.lower(), countycode)) zones[zone]["flash_flood_warning"] = ( "https://tgftp.nws.noaa.gov/data/watches_warnings/" "flash_flood/warning/%s/%s.txt" % (state.lower(), countycode)) zones[zone]["flash_flood_watch"] = ( "https://tgftp.nws.noaa.gov/data/watches_warnings/" "flash_flood/watch/%s/%s.txt" % (state.lower(), zone)) zones[zone]["flood_warning"] = ( "https://tgftp.nws.noaa.gov/data/watches_warnings/" "flood/warning/%s/%s.txt" % (state.lower(), countycode)) zones[zone]["severe_thunderstorm_warning"] = ( "https://tgftp.nws.noaa.gov/data/watches_warnings/" "thunderstorm/%s/%s.txt" % (state.lower(), countycode)) zones[zone]["severe_weather_statement"] = ( "https://tgftp.nws.noaa.gov/data/watches_warnings/" "severe_weather_stmt/%s/%s.txt" % (state.lower(), countycode)) zones[zone]["short_term_forecast"] = ( "https://tgftp.nws.noaa.gov/data/forecasts/nowcast/" "%s/%s.txt" % (state.lower(), zone)) zones[zone]["special_weather_statement"] = ( "https://tgftp.nws.noaa.gov/data/watches_warnings/" "special_weather_stmt/%s/%s.txt" % (state.lower(), zone)) zones[zone]["state_forecast"] = ( "https://tgftp.nws.noaa.gov/data/forecasts/state/" "%s/%s.txt" % (state.lower(), zone)) zones[zone]["tornado"] = ( "https://tgftp.nws.noaa.gov/data/watches_warnings/" "tornado/%s/%s.txt" % (state.lower(), countycode)) zones[zone]["urgent_weather_message"] = ( "https://tgftp.nws.noaa.gov/data/watches_warnings/" "non_precip/%s/%s.txt" % (state.lower(), zone)) zones[zone]["zone_forecast"] = ( "https://tgftp.nws.noaa.gov/data/forecasts/zone/" "%s/%s.txt" % (state.lower(), zone)) tzcode = fields[7] if tzcode == "A": zones[zone]["tz"] = "US/Alaska" elif tzcode == "AH": zones[zone]["tz"] = "US/Aleutian" elif tzcode in ("C", "CE", "CM"): zones[zone]["tz"] = "US/Central" elif tzcode in ("E", "e"): zones[zone]["tz"] = "US/Eastern" elif tzcode == "F": zones[zone]["tz"] = "Pacific/Guadalcanal" elif tzcode == "G": zones[zone]["tz"] = "Pacific/Guam" elif tzcode == "H": zones[zone]["tz"] = "US/Hawaii" elif tzcode == "J": zones[zone]["tz"] = "Japan" elif tzcode == "K": zones[zone]["tz"] = "Pacific/Kwajalein" elif tzcode in ("M", "MC", "MP"): zones[zone]["tz"] = "US/Mountain" elif tzcode == "m": zones[zone]["tz"] = "US/Arizona" elif tzcode == "P": zones[zone]["tz"] = "US/Pacific" elif tzcode == "S": zones[zone]["tz"] = "US/Samoa" elif tzcode == "V": zones[zone]["tz"] = "America/Virgin" else: zones[zone]["tz"] = "" county = fields[5] if county: if description.endswith(county): description += " County" else: description += ", %s County" % county description += ", %s, US" % state zones[zone]["description"] = description zones[zone]["centroid"] = gecos( ",".join( fields[9:11] ) ) if fips in places and not zones[zone]["centroid"]: zones[zone]["centroid"] = places[fips]["centroid"] count += 1 cpfzcf.close() print("done (%s lines)." % count) message = "Reading %s:%s..." % (gzcta_an, gzcta_fn) sys.stdout.write(message) sys.stdout.flush() count = 0 gzcta = zipfile.ZipFile(gzcta_an).open(gzcta_fn, "r") columns = gzcta.readline().decode("utf-8").strip().split("\t") for line in gzcta: fields = line.decode("utf-8").strip().split("\t") f_geoid = fields[ columns.index("GEOID") ].strip() f_intptlat = fields[ columns.index("INTPTLAT") ].strip() f_intptlong = fields[ columns.index("INTPTLONG") ].strip() if f_geoid and f_intptlat and f_intptlong: if f_geoid not in zctas: zctas[f_geoid] = {} zctas[f_geoid]["centroid"] = gecos( "%s,%s" % (f_intptlat, f_intptlong) ) count += 1 gzcta.close() print("done (%s lines)." % count) message = "Reading %s..." % overrides_fn sys.stdout.write(message) sys.stdout.flush() count = 0 added = 0 removed = 0 changed = 0 overrides = configparser.ConfigParser() overrides.read_file( codecs.open(overrides_fn, "r", "utf8") ) overrideslog = [] for section in overrides.sections(): addopt = 0 chgopt = 0 if section.startswith("-"): section = section[1:] delete = True else: delete = False if re.match("[A-Za-z]{3}$", section): if delete: if section in airports: del( airports[section] ) logact = "removed airport %s" % section removed += 1 else: logact = "tried to remove nonexistent airport %s" % section else: if section in airports: logact = "changed airport %s" % section changed += 1 else: airports[section] = {} logact = "added airport %s" % section added += 1 for key,value in overrides.items(section): if key in airports[section]: chgopt += 1 else: addopt += 1 if key in ("centroid", "location"): airports[section][key] = eval(value) else: airports[section][key] = value if addopt and chgopt: logact += " (+%s/!%s options)" % (addopt, chgopt) elif addopt: logact += " (+%s options)" % addopt elif chgopt: logact += " (!%s options)" % chgopt elif re.match("[A-Za-z0-9]{4}$", section): if delete: if section in stations: del( stations[section] ) logact = "removed station %s" % section removed += 1 else: logact = "tried to remove nonexistent station %s" % section else: if section in stations: logact = "changed station %s" % section changed += 1 else: stations[section] = {} logact = "added station %s" % section added += 1 for key,value in overrides.items(section): if key in stations[section]: chgopt += 1 else: addopt += 1 if key in ("centroid", "location"): stations[section][key] = eval(value) else: stations[section][key] = value if addopt and chgopt: logact += " (+%s/!%s options)" % (addopt, chgopt) elif addopt: logact += " (+%s options)" % addopt elif chgopt: logact += " (!%s options)" % chgopt elif re.match("[0-9]{5}$", section): if delete: if section in zctas: del( zctas[section] ) logact = "removed zcta %s" % section removed += 1 else: logact = "tried to remove nonexistent zcta %s" % section else: if section in zctas: logact = "changed zcta %s" % section changed += 1 else: zctas[section] = {} logact = "added zcta %s" % section added += 1 for key,value in overrides.items(section): if key in zctas[section]: chgopt += 1 else: addopt += 1 if key in ("centroid", "location"): zctas[section][key] = eval(value) else: zctas[section][key] = value if addopt and chgopt: logact += " (+%s/!%s options)" % (addopt, chgopt) elif addopt: logact += " (+%s options)" % addopt elif chgopt: logact += " (!%s options)" % chgopt elif re.match("[A-Za-z]{2}[Zz][0-9]{3}$", section): if delete: if section in zones: del( zones[section] ) logact = "removed zone %s" % section removed += 1 else: logact = "tried to remove nonexistent zone %s" % section else: if section in zones: logact = "changed zone %s" % section changed += 1 else: zones[section] = {} logact = "added zone %s" % section added += 1 for key,value in overrides.items(section): if key in zones[section]: chgopt += 1 else: addopt += 1 if key in ("centroid", "location"): zones[section][key] = eval(value) else: zones[section][key] = value if addopt and chgopt: logact += " (+%s/!%s options)" % (addopt, chgopt) elif addopt: logact += " (+%s options)" % addopt elif chgopt: logact += " (!%s options)" % chgopt elif re.match("fips[0-9]+$", section): if delete: if section in places: del( places[section] ) logact = "removed place %s" % section removed += 1 else: logact = "tried to remove nonexistent place %s" % section else: if section in places: logact = "changed place %s" % section changed += 1 else: places[section] = {} logact = "added place %s" % section added += 1 for key,value in overrides.items(section): if key in places[section]: chgopt += 1 else: addopt += 1 if key in ("centroid", "location"): places[section][key] = eval(value) else: places[section][key] = value if addopt and chgopt: logact += " (+%s/!%s options)" % (addopt, chgopt) elif addopt: logact += " (+%s options)" % addopt elif chgopt: logact += " (!%s options)" % chgopt count += 1 overrideslog.append("%s\n" % logact) overrideslog.sort() if os.path.exists(overrideslog_fn): os.rename(overrideslog_fn, "%s_old"%overrideslog_fn) overrideslog_fd = codecs.open(overrideslog_fn, "w", "utf8") import time overrideslog_fd.write( '# Copyright (c) %s Jeremy Stanley . Permission to\n' '# use, copy, modify, and distribute this software is granted under terms\n' '# provided in the LICENSE file distributed with this software.\n\n' % time.gmtime().tm_year) overrideslog_fd.writelines(overrideslog) overrideslog_fd.close() print("done (%s overridden sections: +%s/-%s/!%s)." % ( count, added, removed, changed ) ) estimate = 2*len(places) + len(stations) + 2*len(zctas) + len(zones) print( "Correlating places, stations, ZCTAs and zones (upper bound is %s):" % \ estimate ) count = 0 milestones = list( range(51) ) message = " " sys.stdout.write(message) sys.stdout.flush() for fips in places: centroid = places[fips]["centroid"] if centroid: station = closest(centroid, stations, "location", 0.1) if station[0]: places[fips]["station"] = station count += 1 if not count%100: level = int(50*count/estimate) if level in milestones: for remaining in milestones[:milestones.index(level)+1]: if remaining%5: message = "." sys.stdout.write(message) sys.stdout.flush() else: message = "%s%%" % (remaining*2,) sys.stdout.write(message) sys.stdout.flush() milestones.remove(remaining) if centroid: zone = closest(centroid, zones, "centroid", 0.1) if zone[0]: places[fips]["zone"] = zone count += 1 if not count%100: level = int(50*count/estimate) if level in milestones: for remaining in milestones[:milestones.index(level)+1]: if remaining%5: message = "." sys.stdout.write(message) sys.stdout.flush() else: message = "%s%%" % (remaining*2,) sys.stdout.write(message) sys.stdout.flush() milestones.remove(remaining) for station in stations: if "location" in stations[station]: location = stations[station]["location"] if location: zone = closest(location, zones, "centroid", 0.1) if zone[0]: stations[station]["zone"] = zone count += 1 if not count%100: level = int(50*count/estimate) if level in milestones: for remaining in milestones[:milestones.index(level)+1]: if remaining%5: message = "." sys.stdout.write(message) sys.stdout.flush() else: message = "%s%%" % (remaining*2,) sys.stdout.write(message) sys.stdout.flush() milestones.remove(remaining) for zcta in zctas.keys(): centroid = zctas[zcta]["centroid"] if centroid: station = closest(centroid, stations, "location", 0.1) if station[0]: zctas[zcta]["station"] = station count += 1 if not count%100: level = int(50*count/estimate) if level in milestones: for remaining in milestones[ : milestones.index(level)+1 ]: if remaining%5: message = "." sys.stdout.write(message) sys.stdout.flush() else: message = "%s%%" % (remaining*2,) sys.stdout.write(message) sys.stdout.flush() milestones.remove(remaining) if centroid: zone = closest(centroid, zones, "centroid", 0.1) if zone[0]: zctas[zcta]["zone"] = zone count += 1 if not count%100: level = int(50*count/estimate) if level in milestones: for remaining in milestones[:milestones.index(level)+1]: if remaining%5: message = "." sys.stdout.write(message) sys.stdout.flush() else: message = "%s%%" % (remaining*2,) sys.stdout.write(message) sys.stdout.flush() milestones.remove(remaining) for zone in zones.keys(): if "centroid" in zones[zone]: centroid = zones[zone]["centroid"] if centroid: station = closest(centroid, stations, "location", 0.1) if station[0]: zones[zone]["station"] = station count += 1 if not count%100: level = int(50*count/estimate) if level in milestones: for remaining in milestones[:milestones.index(level)+1]: if remaining%5: message = "." sys.stdout.write(message) sys.stdout.flush() else: message = "%s%%" % (remaining*2,) sys.stdout.write(message) sys.stdout.flush() milestones.remove(remaining) for remaining in milestones: if remaining%5: message = "." sys.stdout.write(message) sys.stdout.flush() else: message = "%s%%" % (remaining*2,) sys.stdout.write(message) sys.stdout.flush() print("\n done (%s correlations)." % count) message = "Writing %s..." % airports_fn sys.stdout.write(message) sys.stdout.flush() count = 0 if os.path.exists(airports_fn): os.rename(airports_fn, "%s_old"%airports_fn) airports_fd = codecs.open(airports_fn, "w", "utf8") airports_fd.write(header) for airport in sorted( airports.keys() ): airports_fd.write("\n\n[%s]" % airport) for key, value in sorted( airports[airport].items() ): if type(value) is float: value = "%.7f"%value elif type(value) is tuple: elements = [] for element in value: if type(element) is float: elements.append("%.7f"%element) else: elements.append( repr(element) ) value = "(%s)"%", ".join(elements) airports_fd.write( "\n%s = %s" % (key, value) ) count += 1 airports_fd.write("\n") airports_fd.close() print("done (%s sections)." % count) message = "Writing %s..." % places_fn sys.stdout.write(message) sys.stdout.flush() count = 0 if os.path.exists(places_fn): os.rename(places_fn, "%s_old"%places_fn) places_fd = codecs.open(places_fn, "w", "utf8") places_fd.write(header) for fips in sorted( places.keys() ): places_fd.write("\n\n[%s]" % fips) for key, value in sorted( places[fips].items() ): if type(value) is float: value = "%.7f"%value elif type(value) is tuple: elements = [] for element in value: if type(element) is float: elements.append("%.7f"%element) else: elements.append( repr(element) ) value = "(%s)"%", ".join(elements) places_fd.write( "\n%s = %s" % (key, value) ) count += 1 places_fd.write("\n") places_fd.close() print("done (%s sections)." % count) message = "Writing %s..." % stations_fn sys.stdout.write(message) sys.stdout.flush() count = 0 if os.path.exists(stations_fn): os.rename(stations_fn, "%s_old"%stations_fn) stations_fd = codecs.open(stations_fn, "w", "utf-8") stations_fd.write(header) for station in sorted( stations.keys() ): stations_fd.write("\n\n[%s]" % station) for key, value in sorted( stations[station].items() ): if type(value) is float: value = "%.7f"%value elif type(value) is tuple: elements = [] for element in value: if type(element) is float: elements.append("%.7f"%element) else: elements.append( repr(element) ) value = "(%s)"%", ".join(elements) if type(value) is bytes: value = value.decode("utf-8") stations_fd.write( "\n%s = %s" % (key, value) ) count += 1 stations_fd.write("\n") stations_fd.close() print("done (%s sections)." % count) message = "Writing %s..." % zctas_fn sys.stdout.write(message) sys.stdout.flush() count = 0 if os.path.exists(zctas_fn): os.rename(zctas_fn, "%s_old"%zctas_fn) zctas_fd = codecs.open(zctas_fn, "w", "utf8") zctas_fd.write(header) for zcta in sorted( zctas.keys() ): zctas_fd.write("\n\n[%s]" % zcta) for key, value in sorted( zctas[zcta].items() ): if type(value) is float: value = "%.7f"%value elif type(value) is tuple: elements = [] for element in value: if type(element) is float: elements.append("%.7f"%element) else: elements.append( repr(element) ) value = "(%s)"%", ".join(elements) zctas_fd.write( "\n%s = %s" % (key, value) ) count += 1 zctas_fd.write("\n") zctas_fd.close() print("done (%s sections)." % count) message = "Writing %s..." % zones_fn sys.stdout.write(message) sys.stdout.flush() count = 0 if os.path.exists(zones_fn): os.rename(zones_fn, "%s_old"%zones_fn) zones_fd = codecs.open(zones_fn, "w", "utf8") zones_fd.write(header) for zone in sorted( zones.keys() ): zones_fd.write("\n\n[%s]" % zone) for key, value in sorted( zones[zone].items() ): if type(value) is float: value = "%.7f"%value elif type(value) is tuple: elements = [] for element in value: if type(element) is float: elements.append("%.7f"%element) else: elements.append( repr(element) ) value = "(%s)"%", ".join(elements) zones_fd.write( "\n%s = %s" % (key, value) ) count += 1 zones_fd.write("\n") zones_fd.close() print("done (%s sections)." % count) message = "Starting QA check..." sys.stdout.write(message) sys.stdout.flush() airports = configparser.ConfigParser() airports.read(airports_fn, encoding="utf-8") places = configparser.ConfigParser() places.read(places_fn, encoding="utf-8") stations = configparser.ConfigParser() stations.read(stations_fn, encoding="utf-8") zctas = configparser.ConfigParser() zctas.read(zctas_fn, encoding="utf-8") zones = configparser.ConfigParser() zones.read(zones_fn, encoding="utf-8") qalog = [] places_nocentroid = 0 places_nodescription = 0 for place in sorted( places.sections() ): if not places.has_option(place, "centroid"): qalog.append("%s: no centroid\n" % place) places_nocentroid += 1 if not places.has_option(place, "description"): qalog.append("%s: no description\n" % place) places_nodescription += 1 stations_nodescription = 0 stations_nolocation = 0 stations_nometar = 0 for station in sorted( stations.sections() ): if not stations.has_option(station, "description"): qalog.append("%s: no description\n" % station) stations_nodescription += 1 if not stations.has_option(station, "location"): qalog.append("%s: no location\n" % station) stations_nolocation += 1 if not stations.has_option(station, "metar"): qalog.append("%s: no metar\n" % station) stations_nometar += 1 airports_badstation = 0 airports_nostation = 0 for airport in sorted( airports.sections() ): if not airports.has_option(airport, "station"): qalog.append("%s: no station\n" % airport) airports_nostation += 1 else: station = airports.get(airport, "station") if station not in stations.sections(): qalog.append( "%s: bad station %s\n" % (airport, station) ) airports_badstation += 1 zctas_nocentroid = 0 for zcta in sorted( zctas.sections() ): if not zctas.has_option(zcta, "centroid"): qalog.append("%s: no centroid\n" % zcta) zctas_nocentroid += 1 zones_nocentroid = 0 zones_nodescription = 0 zones_notz = 0 zones_noforecast = 0 zones_overlapping = 0 zonetable = {} for zone in zones.sections(): if zones.has_option(zone, "centroid"): zonetable[zone] = { "centroid": eval( zones.get(zone, "centroid") ) } for zone in sorted( zones.sections() ): if zones.has_option(zone, "centroid"): zonetable_local = zonetable.copy() del( zonetable_local[zone] ) centroid = eval( zones.get(zone, "centroid") ) if centroid: nearest = closest(centroid, zonetable_local, "centroid", 0.1) if nearest[1]*radian_to_km < 1: qalog.append( "%s: within one km of %s\n" % ( zone, nearest[0] ) ) zones_overlapping += 1 else: qalog.append("%s: no centroid\n" % zone) zones_nocentroid += 1 if not zones.has_option(zone, "description"): qalog.append("%s: no description\n" % zone) zones_nodescription += 1 if not zones.has_option(zone, "tz") or not zones.get( zone, "tz") in zoneinfo.available_timezones(): qalog.append("%s: no time zone\n" % zone) zones_notz += 1 if not zones.has_option(zone, "zone_forecast"): qalog.append("%s: no forecast\n" % zone) zones_noforecast += 1 if os.path.exists(qalog_fn): os.rename(qalog_fn, "%s_old"%qalog_fn) qalog_fd = codecs.open(qalog_fn, "w", "utf8") import time qalog_fd.write( '# Copyright (c) %s Jeremy Stanley . Permission to\n' '# use, copy, modify, and distribute this software is granted under terms\n' '# provided in the LICENSE file distributed with this software.\n\n' % time.gmtime().tm_year) qalog_fd.writelines(qalog) qalog_fd.close() if qalog: print("issues found (see %s for details):"%qalog_fn) if airports_badstation: print(" %s airports with invalid station"%airports_badstation) if airports_nostation: print(" %s airports with no station"%airports_nostation) if places_nocentroid: print(" %s places with no centroid"%places_nocentroid) if places_nodescription: print(" %s places with no description"%places_nodescription) if stations_nodescription: print(" %s stations with no description"%stations_nodescription) if stations_nolocation: print(" %s stations with no location"%stations_nolocation) if stations_nometar: print(" %s stations with no METAR"%stations_nometar) if zctas_nocentroid: print(" %s ZCTAs with no centroid"%zctas_nocentroid) if zones_nocentroid: print(" %s zones with no centroid"%zones_nocentroid) if zones_nodescription: print(" %s zones with no description"%zones_nodescription) if zones_notz: print(" %s zones with no time zone"%zones_notz) if zones_noforecast: print(" %s zones with no forecast"%zones_noforecast) if zones_overlapping: print(" %s zones within one km of another"%zones_overlapping) else: print("no issues found.") print("Indexing complete!")