zyphlar
228 lines
8.4 KiB
Python
228 lines
8.4 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Download OSM data from Overpass API for a given county and save as GeoJSON.
|
|
|
|
Usage:
|
|
python download-overpass.py --type highways "Sumter County" "Florida" output/roads.geojson
|
|
python download-overpass.py --type addresses "Lake County" "Florida" output/addresses.geojson
|
|
python download-overpass.py --type multimodal "Sumter County" "Florida" output/paths.geojson
|
|
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
import time
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
|
|
def get_county_area_id(county_name, state_name):
|
|
"""Get OSM area ID for a county using Nominatim."""
|
|
search_query = f"{county_name}, {state_name}, USA"
|
|
url = f"https://nominatim.openstreetmap.org/search?q={urllib.parse.quote(search_query)}&format=json&limit=1&featuretype=county"
|
|
|
|
# Nominatim requires User-Agent header
|
|
req = urllib.request.Request(url, headers={'User-Agent': 'TheVillagesImport/1.0'})
|
|
|
|
try:
|
|
with urllib.request.urlopen(req) as response:
|
|
results = json.loads(response.read().decode("utf-8"))
|
|
|
|
if results and results[0].get('osm_type') == 'relation':
|
|
relation_id = int(results[0]['osm_id'])
|
|
area_id = relation_id + 3600000000
|
|
print(f"Found {county_name}, {state_name}: relation {relation_id} -> area {area_id}")
|
|
return area_id
|
|
|
|
raise ValueError(f"Could not find relation for {county_name}, {state_name}")
|
|
except urllib.error.HTTPError as e:
|
|
print(f"Nominatim HTTP Error {e.code}: {e.reason}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
|
|
def build_overpass_query(county_name, state_name, data_type="highways"):
|
|
"""Build Overpass API query for specified data type in a county."""
|
|
area_id = get_county_area_id(county_name, state_name)
|
|
|
|
# Addresses: include nwr so building ways and relations are captured too,
|
|
# but use "out center;" instead of "out geom;" so ways/relations return a
|
|
# single centroid rather than their full coordinate list — much smaller payload.
|
|
if data_type == "addresses":
|
|
query = f"""[out:json][timeout:180];
|
|
area(id:{area_id})->.searchArea;
|
|
nwr["addr:housenumber"](area.searchArea);
|
|
out center;"""
|
|
return query
|
|
|
|
base_query = f"""[out:json][timeout:180];
|
|
area(id:{area_id})->.searchArea;"""
|
|
|
|
if data_type == "highways":
|
|
selector = '('
|
|
selector += 'way["highway"="motorway"](area.searchArea);'
|
|
selector += 'way["highway"="trunk"](area.searchArea);'
|
|
selector += 'way["highway"="primary"](area.searchArea);'
|
|
selector += 'way["highway"="secondary"](area.searchArea);'
|
|
selector += 'way["highway"="tertiary"](area.searchArea);'
|
|
selector += 'way["highway"="unclassified"](area.searchArea);'
|
|
selector += 'way["highway"="residential"](area.searchArea);'
|
|
selector += 'way["highway"~"_link"](area.searchArea);'
|
|
selector += 'way["highway"="service"](area.searchArea);'
|
|
selector += 'way["highway"="track"](area.searchArea);'
|
|
selector += ')->.ways;'
|
|
elif data_type == "multimodal":
|
|
selector = '(way["highway"="path"](area.searchArea);way["highway"="cycleway"](area.searchArea);)->.ways;'
|
|
else:
|
|
raise ValueError(f"Unknown data type: {data_type}")
|
|
|
|
query = base_query + selector + ".ways out meta geom;\nnode(w.ways);\nout meta;"
|
|
return query
|
|
|
|
|
|
def query_overpass(query):
|
|
"""Send query to Overpass API and return JSON response, with retries."""
|
|
url = "https://overpass-api.de/api/interpreter"
|
|
data = urllib.parse.urlencode({"data": query}).encode("utf-8")
|
|
req = urllib.request.Request(
|
|
url,
|
|
data=data,
|
|
headers={"Accept": "*/*", "User-Agent": "TheVillagesImport/1.0"},
|
|
)
|
|
|
|
max_attempts = 3
|
|
for attempt in range(1, max_attempts + 1):
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=300) as response:
|
|
return json.loads(response.read().decode("utf-8"))
|
|
except urllib.error.HTTPError as e:
|
|
try:
|
|
error_body = e.read().decode("utf-8")
|
|
except Exception:
|
|
error_body = "(unreadable)"
|
|
print(f"HTTP Error {e.code} on attempt {attempt}: {e.reason}", file=sys.stderr)
|
|
print(f"Error response body: {error_body}", file=sys.stderr)
|
|
if attempt < max_attempts and e.code in (429, 504):
|
|
wait = 30 * attempt
|
|
print(f"Retrying in {wait}s...", file=sys.stderr)
|
|
time.sleep(wait)
|
|
else:
|
|
sys.exit(1)
|
|
except Exception as e:
|
|
print(f"Error querying Overpass API: {e}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
|
|
def convert_to_geojson(overpass_data):
|
|
"""Convert Overpass API response to GeoJSON format."""
|
|
# First pass: collect node versions (present when query includes node metadata)
|
|
node_versions = {}
|
|
for element in overpass_data.get("elements", []):
|
|
if element["type"] == "node" and "version" in element:
|
|
node_versions[element["id"]] = element["version"]
|
|
|
|
features = []
|
|
|
|
for element in overpass_data.get("elements", []):
|
|
if element["type"] == "node":
|
|
if not element.get("tags"):
|
|
continue # Skip topology-only nodes (no tags); versions collected above
|
|
feature = {
|
|
"type": "Feature",
|
|
"properties": element.get("tags", {}),
|
|
"geometry": {
|
|
"type": "Point",
|
|
"coordinates": [element["lon"], element["lat"]]
|
|
}
|
|
}
|
|
features.append(feature)
|
|
|
|
elif element["type"] in ("way", "relation"):
|
|
if "geometry" in element:
|
|
# out meta geom; — full coordinate list + node IDs + version
|
|
coordinates = [[coord["lon"], coord["lat"]] for coord in element["geometry"]]
|
|
props = dict(element.get("tags", {}))
|
|
props["osm_id"] = element["id"]
|
|
props["osm_type"] = element["type"]
|
|
props["osm_version"] = element.get("version", 1)
|
|
if "nodes" in element:
|
|
props["osm_nodes"] = element["nodes"]
|
|
props["osm_node_versions"] = [node_versions.get(nid, 1) for nid in element["nodes"]]
|
|
feature = {
|
|
"type": "Feature",
|
|
"properties": props,
|
|
"geometry": {
|
|
"type": "LineString",
|
|
"coordinates": coordinates
|
|
}
|
|
}
|
|
features.append(feature)
|
|
elif "center" in element:
|
|
# out center; — single centroid point (used for addresses on buildings)
|
|
c = element["center"]
|
|
feature = {
|
|
"type": "Feature",
|
|
"properties": element.get("tags", {}),
|
|
"geometry": {
|
|
"type": "Point",
|
|
"coordinates": [c["lon"], c["lat"]]
|
|
}
|
|
}
|
|
features.append(feature)
|
|
|
|
return {
|
|
"type": "FeatureCollection",
|
|
"features": features
|
|
}
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(
|
|
description="Download OSM data from Overpass API for a county"
|
|
)
|
|
parser.add_argument(
|
|
"county",
|
|
help="County name (e.g., 'Lake County')"
|
|
)
|
|
parser.add_argument(
|
|
"state",
|
|
help="State name (e.g., 'Florida')"
|
|
)
|
|
parser.add_argument(
|
|
"output",
|
|
help="Output GeoJSON file path"
|
|
)
|
|
parser.add_argument(
|
|
"--type", "-t",
|
|
choices=["highways", "addresses", "multimodal"],
|
|
default="highways",
|
|
help="Type of data to download (default: highways)"
|
|
)
|
|
|
|
args = parser.parse_args()
|
|
|
|
# Create output directory if it doesn't exist
|
|
output_path = Path(args.output)
|
|
output_path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
print(f"Downloading {args.type} for {args.county}, {args.state}...")
|
|
|
|
# Build and execute query
|
|
query = build_overpass_query(args.county, args.state, args.type)
|
|
print(f"Query: {query}")
|
|
overpass_data = query_overpass(query)
|
|
|
|
# Convert to GeoJSON
|
|
geojson = convert_to_geojson(overpass_data)
|
|
|
|
# Save to file
|
|
with open(output_path, "w", encoding="utf-8") as f:
|
|
json.dump(geojson, f, indent=2)
|
|
|
|
print(f"Saved {len(geojson['features'])} {args.type} features to {args.output}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |