#!/usr/bin/env python3 """ Geocode all asset addresses from the assets DB, cache results, and update GPS. Sources (priority): 1. Existing geocache.json 2. Fresh Nominatim (OpenStreetMap) geocoding — rate-limited to 1/sec Usage: python3 scripts/geocode_asset_addresses.py # dry-run python3 scripts/geocode_asset_addresses.py --apply # write to DB python3 scripts/geocode_asset_addresses.py --force # re-geocode cached too """ import json import sqlite3 import sys import time import urllib.parse import urllib.request from pathlib import Path ASSET_DB = str(Path(__file__).resolve().parent.parent / "assets.db") GEOCODE_CACHE = str( Path.home() / "projects" / "ms-field-service-extraction" / "web" / "static" / "data" / "geocache.json" ) USER_AGENT = "CanteenAssetTracker/1.0 (internal tool; canteen.ourpad.casa)" NOMINATIM_URL = "https://nominatim.openstreetmap.org/search" RATE_LIMIT = 1.1 # seconds between requests def load_geocache(): """Load existing geocache from file.""" if Path(GEOCODE_CACHE).exists(): with open(GEOCODE_CACHE) as f: return json.load(f) return {} def save_geocache(cache): """Save geocache back to file.""" Path(GEOCODE_CACHE).parent.mkdir(parents=True, exist_ok=True) with open(GEOCODE_CACHE, "w") as f: json.dump(cache, f, indent=2) def geocode(address): """Geocode a single address via Nominatim. Returns (lat, lng) or (None, None).""" if not address or not address.strip(): return None, None params = urllib.parse.urlencode({ "q": address.strip(), "format": "json", "limit": 1, "addressdetails": "0", }) url = f"{NOMINATIM_URL}?{params}" req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT}) try: with urllib.request.urlopen(req, timeout=10) as resp: data = json.loads(resp.read()) if data and len(data) > 0: lat = float(data[0]["lat"]) lng = float(data[0]["lon"]) return lat, lng except Exception as e: print(f" ⚠ Error geocoding '{address[:60]}': {e}", file=sys.stderr) return None, None def make_addr_key(asset): """Build a normalized address key from asset record.""" parts = [ asset.get("address", "") or "", asset.get("seed_location", "") or "", ] # Build full address with city/state addr = "" if asset.get("address"): addr = asset["address"].strip() if asset.get("seed_city"): addr += f", {asset['seed_city'].strip()}" if asset.get("state"): addr += f", {asset['state'].strip()}" if asset.get("postal_code"): addr += f", {asset['postal_code'].strip()}" elif asset.get("seed_location"): addr = asset["seed_location"].strip() return addr.strip().lower() if addr else None def build_full_address(asset): """Build the cleanest full address string for geocoding. Uses only the street address + city + state + zip. Excludes seed_location (extra venue/building names confuse Nominatim). """ addr = (asset.get("address") or "").strip() if not addr: addr = (asset.get("seed_location") or "").strip() loc_parts = [] if asset.get("seed_city"): loc_parts.append(asset["seed_city"].strip()) if asset.get("state"): loc_parts.append(asset["state"].strip()) if asset.get("postal_code"): loc_parts.append(asset["postal_code"].strip()) if loc_parts: # Add "FL" if not already present in the state field if "FL" not in loc_parts and "Florida" not in loc_parts: loc_parts.insert(0, "FL") addr += ", " + ", ".join(loc_parts) elif not addr: return None return addr.strip().strip(",").strip() def main(): apply_changes = "--apply" in sys.argv force = "--force" in sys.argv mode = "DRY RUN (no writes)" if not apply_changes else "APPLYING changes" print(f"Assets DB: {ASSET_DB}") print(f"Geocache: {GEOCODE_CACHE}") print(f"Mode: {mode}") if force: print(" (--force: re-geocode even cached addresses)") print() # 1. Load existing cache cache = load_geocache() cached_with_coords = sum(1 for v in cache.values() if v.get("lat")) cached_misses = sum(1 for v in cache.values() if not v.get("lat")) print(f"📦 Geocache: {len(cache)} entries ({cached_with_coords} with coords, {cached_misses} misses)") # 2. Get all assets without GPS but with addresses conn = sqlite3.connect(ASSET_DB) conn.execute("PRAGMA journal_mode=WAL") cur = conn.cursor() cur.execute(""" SELECT machine_id, name, address, seed_location, seed_city, state, postal_code FROM assets WHERE latitude IS NULL AND ((address IS NOT NULL AND address != '') OR (seed_location IS NOT NULL AND seed_location != '')) ORDER BY machine_id """) rows = cur.fetchall() print(f"📋 Assets needing geocoding: {len(rows)}") if not rows: print("✅ No assets need geocoding. Done.") return # 3. Build (addr_key -> machine_ids) map for unique addresses addr_to_mids = {} addr_to_full = {} asset_cols = ["machine_id", "name", "address", "seed_location", "seed_city", "state", "postal_code"] col = dict(zip(asset_cols, range(len(asset_cols)))) for row in rows: asset = {k: row[col[k]] for k in asset_cols} addr_key = make_addr_key(asset) if not addr_key: continue if addr_key not in addr_to_mids: addr_to_mids[addr_key] = [] addr_to_full[addr_key] = build_full_address(asset) addr_to_mids[addr_key].append(row[col["machine_id"]]) print(f"📍 Unique addresses to resolve: {len(addr_to_mids)}") # 4. Resolve each unique address resolved = {} # addr_key -> (lat, lon) still_needed = 0 fresh_calls = 0 for addr_key, mids in sorted(addr_to_mids.items()): # Check cache first cached = cache.get(addr_key, {}) if cached and not force: lat, lng = cached.get("lat"), cached.get("lng") if lat and lng: resolved[addr_key] = (lat, lng) continue elif cached.get("lat") is None: # previously cached as miss still_needed += 1 continue # Need to geocode fresh full_addr = addr_to_full[addr_key] print(f" 🌐 Geocoding [{fresh_calls+1}]: {full_addr[:70]}...") lat, lng = geocode(full_addr) # Save to cache cache[addr_key] = {"address": full_addr, "lat": lat, "lng": lng} save_geocache(cache) if lat and lng: resolved[addr_key] = (lat, lng) print(f" ✅ ({lat:.5f}, {lng:.5f}) — affects {len(mids)} assets") else: print(f" ❌ No result — affects {len(mids)} assets") still_needed += 1 fresh_calls += 1 if fresh_calls >= 3 and not apply_changes: # In dry-run mode, only do a few to show what's possible print(f"\n ... stopping after {fresh_calls} (dry-run). Use --apply to geocode all ~{len(addr_to_mids)} addresses.") break time.sleep(RATE_LIMIT) # 5. Count updates total_updatable = 0 updates = [] for addr_key, mids in addr_to_mids.items(): if addr_key in resolved: lat, lon = resolved[addr_key] for mid in mids: updates.append((lat, lon, "geocoded", mid)) total_updatable += len(mids) updates_by_source = sum(len(m) for a, m in addr_to_mids.items() if a in resolved) print(f"\n{'=' * 60}") print(f"📊 SUMMARY") print(f"{'=' * 60}") print(f" Total assets needing GPS: {len(rows)}") print(f" Unique addresses to geocode: {len(addr_to_mids)}") print(f" Resolved from cache/source: {len(resolved) + (cached_with_coords - sum(1 for k in resolved if k not in cache))}") print(f" Freshly geocoded: {fresh_calls}") print(f" Still unresolved: {still_needed}") print(f" Assets that can be updated: {updates_by_source}") if apply_changes and updates: # Apply in batches for speed conn.executemany( "UPDATE assets SET latitude=?, longitude=?, location_source=?, updated_at=datetime('now') WHERE machine_id=?", [(lat, lon, src, mid) for lat, lon, src, mid in updates], ) conn.commit() print(f"\n ✅ Applied {len(updates)} GPS updates from geocoding") elif updates: print(f"\n Run with --apply to write {len(updates)} GPS updates to DB.") # Final count cur = conn.execute("SELECT COUNT(*) FROM assets WHERE latitude IS NOT NULL") final_gps = cur.fetchone()[0] cur = conn.execute("SELECT COUNT(*) FROM assets") total = cur.fetchone()[0] print(f"\n Final: {final_gps} / {total} assets have GPS coordinates") conn.close() if __name__ == "__main__": main()