mirror of
https://github.com/ciphervance/supercell-wx.git
synced 2025-10-30 21:30:05 +00:00
312 lines
12 KiB
Python
312 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
import requests
|
|
import json
|
|
import argparse
|
|
import csv
|
|
|
|
NOAA_BASE = "https://www.ncdc.noaa.gov/homr/services/station"
|
|
WARNING = "\033[93mWARNING: Updating radar sites may break tests in \
|
|
'test/source/scwx/qt/config/radar_site.test.cpp'\033[39m"
|
|
|
|
# Get the noaa station data.
|
|
# platform is what platform should be searched for
|
|
# current (should) filter to only current stations (always is filtered)
|
|
# icao is a ICAO identifier. Without it we search for all.
|
|
def get_noaa_stations(platform, current, icao = None):
|
|
params = {
|
|
"definitions": "false",
|
|
"phrData": "false",
|
|
}
|
|
if platform is not None:
|
|
params["platform"] = platform
|
|
if current:
|
|
params["current"] = "true"
|
|
if icao is not None:
|
|
params["qid"] = f"ICAO:{icao}"
|
|
|
|
res = requests.get(NOAA_BASE + "/search", params = params)
|
|
|
|
if res.ok:
|
|
return res.json()["stationCollection"]["stations"]
|
|
else:
|
|
print("NETWORK ERROR: Could not get resources from NOAA HOMR")
|
|
print(res.text)
|
|
exit(5)
|
|
|
|
# dictionary to convert NOAA types to Supercell_wx types
|
|
NOAA_TYPE_DICT = {
|
|
"TDWR": "tdwr",
|
|
"NEXRAD": "wsr88d"
|
|
}
|
|
|
|
# Given an list of objects, find the object with the best value for key.
|
|
# The values that appear earlier in values are better.
|
|
# subKey will take return the value under that key, not the full object.
|
|
# parser (needs subKey) is a function that will have the value found by
|
|
# subKey and return a parsed version of it (often 'float' because HOMR
|
|
# data uses strings for floats)
|
|
def extract_best(items, key, values, subKey = None, parser = None):
|
|
valuesPart = enumerate(reversed(values))
|
|
valueDict = dict([(k,v) for v,k in valuesPart])
|
|
best = None
|
|
bestInd = -1
|
|
|
|
for item in items:
|
|
index = valueDict.get(item[key], -1)
|
|
if bestInd is None or bestInd < index:
|
|
bestInd = index
|
|
best = item
|
|
|
|
|
|
if subKey is None or best is None:
|
|
return best
|
|
|
|
if parser is not None:
|
|
return parser(best[subKey])
|
|
|
|
return best[subKey]
|
|
|
|
# Turn the noaa stations into a dictionary that can be used to update the locations and elevations.
|
|
def make_noaa_stations_dict(noaaStations):
|
|
output = {}
|
|
|
|
for station in noaaStations:
|
|
stationId = extract_best(station["identifiers"], "idType", ["NEXRAD", "ICAO"], "id")
|
|
|
|
if stationId in output: # some stations are repeaded in non NEXRAD/TDWR locations.
|
|
continue
|
|
|
|
stationDict = {}
|
|
stationDict["lat"] = float(station["header"]["latitude_dec"])
|
|
stationDict["lon"] = float(station["header"]["longitude_dec"])
|
|
stationDict["elevation"] = extract_best(station["location"].get("elevations", []),
|
|
"elevationType",
|
|
["GROUND"],
|
|
"elevationFeet",
|
|
float)
|
|
# These are some things that could be updated from the NOAA HOMR data,
|
|
# but are not necessary in the same format, so they are disabled.
|
|
"""
|
|
stationDict["id"] = stationId
|
|
stationDict["country"] = station["location"]["geoInfo"]["countries"][0]["country"]
|
|
if "stateProvinces" in station["location"]["geoInfo"]:
|
|
stationDict["state"] = station["location"]["geoInfo"]["stateProvinces"][0]["stateProvince"]
|
|
else:
|
|
stationDict["state"] = None
|
|
stationDict["place"] = extract_best(station["names"], "nameType", ["PRINCIPAL"], "name")
|
|
stationDict["type"] = NOAA_TYPE_DICT[station["platforms"][0]["platform"]]
|
|
#stationDict["tz"] = station["location"]["geoInfo"]["utcOffsets"][0]["utcOffset"] # This is UTC offset, not timezone
|
|
"""
|
|
|
|
output[stationId] = stationDict
|
|
|
|
return output
|
|
|
|
# Get the list of updated stations (not in place), using the noaaStationsDict
|
|
# from make_noaa_stations_dict
|
|
def update_stations(noaaStationsDict, previousStations, toUpdate = None):
|
|
newStations = []
|
|
for station in previousStations:
|
|
newStation = station.copy()
|
|
|
|
if not station["id"] in noaaStationsDict:
|
|
# may be good idea to add fallback to a ICAO search for non active
|
|
# stations.
|
|
print(f"WARNING: Station '{station['id']}' not found in noaa data")
|
|
|
|
if "elevation" not in station:
|
|
newStation["elevation"] = None
|
|
elif toUpdate is None or station["id"] in toUpdate:
|
|
newStation.update(noaaStationsDict[station["id"]])
|
|
else:
|
|
newStation["elevation"] = noaaStationsDict[station["id"]]["elevation"]
|
|
|
|
newStations.append(newStation)
|
|
|
|
return newStations
|
|
|
|
# Read in csv file describing which locations to update.
|
|
# Elevation data is always updated.
|
|
def get_to_update_file(filename):
|
|
with open(filename) as file:
|
|
r = csv.reader(file)
|
|
next(r)
|
|
toUpdate = set()
|
|
for row in r:
|
|
if len(row) == 2 and row[1] == "HOMR":
|
|
toUpdate.add(row[0])
|
|
return toUpdate
|
|
|
|
|
|
# Customized dump routine. Formats it as one station per row, aligning items.
|
|
def custom_dump(stations, file):
|
|
file.write("[\n")
|
|
lengths = {}
|
|
lastKey = None
|
|
keys = None
|
|
|
|
# Find length for each value, and ensure all stations have the same keys.
|
|
for station in stations:
|
|
for key, value in station.items():
|
|
length = len(json.dumps(value))
|
|
if key in lengths:
|
|
lengths[key] = max(length, lengths[key])
|
|
else:
|
|
lengths[key] = length
|
|
lastKey = key
|
|
|
|
newKeys = list(station.keys())
|
|
if keys is None:
|
|
keys = newKeys
|
|
elif keys != newKeys:
|
|
print("DUMP ERROR: Stations did not have the same keys.")
|
|
exit(3)
|
|
|
|
# Write out each station with the correct format.
|
|
lastType = None
|
|
for station in stations:
|
|
# put an empty line between NEXRAD and TDWR.
|
|
if lastType is not None and lastType != station["type"]:
|
|
file.write("\n")
|
|
|
|
file.write("\t{ ")
|
|
|
|
for key, value in station.items():
|
|
value = json.dumps(value)
|
|
file.write(f'"{key}": {value}')
|
|
|
|
if key != lastKey:
|
|
file.write(", ")
|
|
file.write(" " * (lengths[key] - len(value)))
|
|
|
|
if station == stations[-1]:
|
|
file.write(" }\n")
|
|
else:
|
|
file.write(" },\n")
|
|
|
|
lastType = station["type"]
|
|
|
|
file.write("]\n")
|
|
|
|
# Write coordinates out to a file. Useful for checking against map program.
|
|
def make_coords(stations, file):
|
|
for station in stations:
|
|
lat = str(abs(station["lat"]))
|
|
lat += "N" if station["lat"] > 0 else "S"
|
|
|
|
lon = str(abs(station["lon"]))
|
|
lon += "E" if station["lon"] > 0 else "W"
|
|
|
|
file.write(f"{lat} {lon}\n")
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(
|
|
description="""Update supercell-wx's location data for towers form NOAA's HOMR database.\n
|
|
Recommended Arguments: -u ../res/config/radar_sites.json -t -w""")
|
|
parser.add_argument("--current_file", "-u", type = str, default = None, required = False,
|
|
help = "The 'radar_sites.json' file to update. Without this option, this will generate a new file")
|
|
parser.add_argument("--test_updated", "-t", default = False, action = "store_true",
|
|
help = "Read in the updated file to ensure it is valid JSON. Should be used.")
|
|
parser.add_argument("--to_update_csv", "-U", type = str, default = None, required = False,
|
|
help = "Choose a CSV describing which stations to update. \
|
|
First column is station ID, Second is HOMR if it should be updated. First row is a header.")
|
|
parser.add_argument("--updated_file", "-o", type = str, default = None, required = False,
|
|
help = "The updated 'radar_sites.json' file. The default is to overwrite the current one.")
|
|
parser.add_argument("--coord_file", "-c", type = str, default = None, required = False,
|
|
help = "Output an additional file with the coordinates of each site.")
|
|
parser.add_argument("--resp_file", "-r", type = str, default = None, required = False,
|
|
help = "Output most of the JSON from the responses.")
|
|
parser.add_argument("--input_json", "-i", type = str, default = None, required = False,
|
|
help = "Instead of querying NOAA, just read in a JSON file made by \"-r\".")
|
|
parser.add_argument("--json_dump", "-j", default = False, action = "store_true",
|
|
help = "Uses 'json.dump' instead of the custom dump function. Has worse formatting.")
|
|
parser.add_argument("--more_radars", "-m", default = False, action = "store_true",
|
|
help = "Get AWOS and UPPERAIR stations as well. Should NOT be used.")
|
|
parser.add_argument("--current_only", "-C", default = False, action = "store_true",
|
|
help = "Get only currently active stations. Does not seem to change anything.")
|
|
parser.add_argument("--warn", "-w", default = False, action = "store_true",
|
|
help = "Display a warning about breaking a test by updating the radar sites.")
|
|
|
|
args = parser.parse_args()
|
|
# default to updating the same file as input
|
|
if args.updated_file is None:
|
|
if args.current_file is None:
|
|
parser.error("Needs 'current_file' or 'updated_file'")
|
|
args.updated_file = args.current_file
|
|
|
|
previousStations = None
|
|
if args.current_file is not None:
|
|
print(f"Reading Current Sites from '{args.current_file}'")
|
|
with open(args.current_file, "r") as file:
|
|
previousStations = json.load(file)
|
|
|
|
toUpdate = None
|
|
if args.to_update_csv is not None:
|
|
toUpdate = get_to_update_file(args.to_update_csv)
|
|
|
|
if args.input_json is None:
|
|
print("Getting NEXRAD stations")
|
|
noaaStations = get_noaa_stations("NEXRAD", args.current_only)
|
|
|
|
print("Getting TDWR stations")
|
|
noaaStations += get_noaa_stations("TDWR", args.current_only)
|
|
|
|
if args.more_radars: # Should not be used
|
|
print("Getting AWOS stations")
|
|
noaaStations += get_noaa_stations("AWOS", args.current_only)
|
|
|
|
print("Getting UPPERAIR stations")
|
|
noaaStations += get_noaa_stations("UPPERAIR", args.current_only)
|
|
else:
|
|
with open(args.input_json, "r") as file:
|
|
noaaStations = json.load(file)
|
|
|
|
if args.resp_file is not None:
|
|
with open(args.resp_file, "w") as file:
|
|
json.dump(noaaStations, file, indent=4)
|
|
|
|
print("Processing Data")
|
|
noaaStationsDict = make_noaa_stations_dict(noaaStations)
|
|
|
|
if args.current_file is None:
|
|
newStations = list(noaaStationsDict.values())
|
|
else:
|
|
newStations = update_stations(noaaStationsDict, previousStations, toUpdate)
|
|
|
|
print(f"Saving Updated Sites to '{args.updated_file}'")
|
|
with open(args.updated_file, "w") as file:
|
|
if args.json_dump:
|
|
json.dump(newStation, file)
|
|
else:
|
|
custom_dump(newStations, file)
|
|
|
|
if args.coord_file is not None:
|
|
print(f"Saving Coordinates to '{args.coord_file}'")
|
|
with open(args.coord_file, "w") as file:
|
|
make_coords(newStations, file)
|
|
|
|
if args.test_updated:
|
|
failed = False
|
|
with open(args.updated_file, "r") as file:
|
|
try:
|
|
data = json.load(file)
|
|
if len(data) < len(newStations):
|
|
print(f"TEST ERROR: Only read in {len(data)} out of {len(newStations)} items.")
|
|
failed = True
|
|
if json.dumps(data) != json.dumps(newStations):
|
|
print(f"TEST ERROR: Dumps are not equal")
|
|
failed = True
|
|
except Exception as e:
|
|
print(e)
|
|
failed = True
|
|
|
|
if failed:
|
|
exit(4)
|
|
|
|
if args.warn:
|
|
print(WARNING)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|
|
|
|
|