the-villages-import/download-overpass.py

#!/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

TODO:
- Don't just download roads. Probably ignore relations also.
"""

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)

    base_query = f"""[out:json][timeout:60];
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 == "addresses":
        selector = 'nwr["addr:housenumber"](area.searchArea);'
    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."""
    url = "https://overpass-api.de/api/interpreter"
    data = urllib.parse.urlencode({"data": query}).encode("utf-8")
    
    
    try:
        with urllib.request.urlopen(url, data=data) as response:
            return json.loads(response.read().decode("utf-8"))
    except urllib.error.HTTPError as e:
        print(f"HTTP Error {e.code}: {e.reason}", file=sys.stderr)
        try:
            error_body = e.read().decode("utf-8")
            print(f"Error response body: {error_body}", file=sys.stderr)
        except:
            print("Could not read error response body", file=sys.stderr)
        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"] == "way" and "geometry" in element:
            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 element["type"] == "node":
            feature = {
                "type": "Feature", 
                "properties": element.get("tags", {}),
                "geometry": {
                    "type": "Point",
                    "coordinates": [element["lon"], element["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()