#!/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:120]; 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 += ');' elif data_type == "multimodal": selector = '(way["highway"="path"](area.searchArea);way["highway"="cycleway"](area.searchArea););' else: raise ValueError(f"Unknown data type: {data_type}") query = base_query + selector + "out geom;" 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") max_attempts = 3 for attempt in range(1, max_attempts + 1): try: with urllib.request.urlopen(url, data=data, 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.""" features = [] for element in overpass_data.get("elements", []): if element["type"] == "node": 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 geom; — full coordinate list (used for highways/paths) coordinates = [[coord["lon"], coord["lat"]] for coord in element["geometry"]] feature = { "type": "Feature", "properties": element.get("tags", {}), "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()