will/osm-import-tools
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

job folders, selection

Browse Source
This commit is contained in:
zyphlar
2026-02-25 10:25:48 -08:00
parent d617750a46
commit de7789442d
5 changed files with 975 additions and 217 deletions
Download Patch File Download Diff File Expand all files Collapse all files
compare-addresses.py
+72 -28
View File
@@ -14,6 +14,7 @@ Usage:
import argparse
import json
import os
import shutil
import sys
import zipfile
from datetime import datetime
@@ -39,10 +40,10 @@ warnings.filterwarnings('ignore')
class AddressComparator:
def __init__(self, tolerance_meters: float = 50.0, cache_dir: str = "osm_cache"):
def __init__(self, tolerance_meters: float = 500.0, cache_dir: str = "osm_cache"):
"""
Initialize the address comparator.
Args:
tolerance_meters: Distance tolerance for considering addresses as matching
cache_dir: Directory to cache OSM data
@@ -50,32 +51,68 @@ class AddressComparator:
self.tolerance_meters = tolerance_meters
self.cache_dir = Path(cache_dir)
self.cache_dir.mkdir(exist_ok=True)
# Convert meters to degrees (approximate)
# 1 degree latitude ≈ 111,000 meters
self.tolerance_deg = tolerance_meters / 111000.0
def download_osm_addresses(self, county: str, state: str, output_file: str = None) -> str:
def _get_county_area_id(self, county: str, state: str) -> int:
"""Get OSM area ID for a county using Nominatim."""
search_query = f"{county} County, {state}, 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} County, {state}: relation {relation_id} -> area {area_id}")
return area_id
raise ValueError(f"Could not find relation for {county} County, {state}")
except urllib.error.HTTPError as e:
print(f"Nominatim HTTP Error {e.code}: {e.reason}", file=sys.stderr)
sys.exit(1)
def download_osm_addresses(self, county: str, state: str, output_file: str = None, output_dir: str = None) -> str:
"""Download address data from OpenStreetMap via Overpass API."""
if output_file is None:
timestamp = datetime.now().strftime("%Y%m%d")
output_file = self.cache_dir / f"osm_addresses_{county.lower()}_{timestamp}.geojson"
# Determine cache file location (with timestamp for caching)
timestamp = datetime.now().strftime("%Y%m%d")
cache_file = self.cache_dir / f"osm_addresses_{county.lower()}_{timestamp}.geojson"
# Determine final output file location (standard name for web serving)
if output_file is not None:
final_output_file = Path(output_file)
elif output_dir is not None:
final_output_file = Path(output_dir) / "osm-addresses.geojson"
else:
output_file = Path(output_file)
final_output_file = cache_file
# Check if cached file exists and is recent (less than 7 days old)
if output_file.exists():
file_age = datetime.now().timestamp() - output_file.stat().st_mtime
if cache_file.exists():
file_age = datetime.now().timestamp() - cache_file.stat().st_mtime
if file_age < 7 * 24 * 3600: # 7 days in seconds
print(f"Using cached OSM data: {output_file}")
return str(output_file)
print(f"Using cached OSM data: {cache_file}")
# Copy cache to final output location if different
if final_output_file != cache_file:
final_output_file.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(cache_file, final_output_file)
print(f"Copied to: {final_output_file}")
return str(final_output_file)
print(f"Downloading OSM addresses for {county} County, {state}...")
# Build Overpass query for addresses
# Get the specific area ID for this county
area_id = self._get_county_area_id(county, state)
# Build Overpass query for addresses using area ID
query = f"""[out:json][timeout:180];
area["name"="{state}"]->.state;
area["name"="{county} County"](area.state)->.searchArea;
area(id:{area_id})->.searchArea;
nwr["addr:housenumber"](area.searchArea);
out geom;"""
@@ -84,14 +121,21 @@ out geom;"""
# Convert to GeoJSON
geojson = self._convert_osm_to_geojson(osm_data)
# Save to file
output_file.parent.mkdir(parents=True, exist_ok=True)
with open(output_file, 'w', encoding='utf-8') as f:
# Save to cache file
cache_file.parent.mkdir(parents=True, exist_ok=True)
with open(cache_file, 'w', encoding='utf-8') as f:
json.dump(geojson, f, indent=2)
print(f"Downloaded {len(geojson['features'])} OSM addresses to {output_file}")
return str(output_file)
print(f"Downloaded {len(geojson['features'])} OSM addresses to cache: {cache_file}")
# Also save to final output location if different
if final_output_file != cache_file:
final_output_file.parent.mkdir(parents=True, exist_ok=True)
with open(final_output_file, 'w', encoding='utf-8') as f:
json.dump(geojson, f, indent=2)
print(f"Saved to: {final_output_file}")
return str(final_output_file)
def _query_overpass(self, query: str) -> Dict[str, Any]:
"""Send query to Overpass API and return JSON response."""
@@ -581,8 +625,8 @@ Examples:
parser.add_argument('county', help='County name (e.g., "Lake", "Sumter")')
parser.add_argument('state', help='State name (e.g., "Florida")')
parser.add_argument('--local-zip', required=True, help='Path to local address data ZIP file')
parser.add_argument('--tolerance', '-t', type=float, default=50.0,
help='Distance tolerance in meters for matching addresses (default: 50)')
parser.add_argument('--tolerance', '-t', type=float, default=500.0,
help='Distance tolerance in meters for matching addresses (default: 500)')
parser.add_argument('--output-dir', '-o', help='Output directory for results (default: processed data/[County])')
parser.add_argument('--cache-dir', default='osm_cache',
help='Directory to cache OSM downloads (default: osm_cache)')
@@ -625,8 +669,8 @@ Examples:
# Remove existing cache for this county
for cache_file in Path(args.cache_dir).glob(f"osm_addresses_{args.county.lower()}_*.geojson"):
cache_file.unlink()
osm_file = comparator.download_osm_addresses(args.county, args.state)
osm_file = comparator.download_osm_addresses(args.county, args.state, output_dir=str(output_dir))
# Convert local data
local_file = comparator.load_local_addresses(args.local_zip)