"""Utilities for deriving geospatial context and reverse geocoding.""" from __future__ import annotations import math from typing import Any, Dict, Iterable, List, Optional, Tuple import requests from shapely.geometry import LineString, MultiPolygon, Point, Polygon from shapely.ops import nearest_points from scipy.signal import convolve2d from rasterio.crs import CRS from rasterio.transform import xy from pyproj import Transformer import numpy as np import os from dotenv import load_dotenv load_dotenv() NOMINATIM_URL = "https://nominatim.openstreetmap.org/reverse" def pixel_to_wgs84(crs, transform, pixel_row, pixel_col): """Convert a raster pixel row/col into a WGS84 (lon, lat) coordinate.""" row, col = pixel_row, pixel_col x, y = xy(transform, row, col, offset="center") source_crs = CRS.from_user_input(crs) target_crs = CRS.from_epsg(4326) if source_crs == target_crs: lon, lat = x, y else: transformer = Transformer.from_crs(source_crs, target_crs, always_xy=True) lon, lat = transformer.transform(x, y) return float(lon), float(lat) def extreme_mean_patches(arr, k=10, n=1, order = "coolest", print_coords = False): """Find the coolest/hottest kxk patches and return their centers with mean values.""" kernel = np.ones((k, k), dtype=float) # Mask invalid windows (containing NaNs) nan_count = convolve2d(np.isnan(arr).astype(float), kernel, mode='valid') valid = nan_count == 0 # Compute mean map sum_map = convolve2d(np.nan_to_num(arr, nan=0.0), kernel, mode='valid') mean_map = np.where(valid, sum_map / (k * k), np.nan) # Flatten and sort only valid means flat_means = mean_map.flatten() valid_indices = np.flatnonzero(np.isfinite(flat_means)) valid_indices_sorted = np.argsort(flat_means[valid_indices]) if order !='coolest':#i.e hottest valid_indices_sorted = valid_indices_sorted[::-1] sorted_idx = valid_indices[valid_indices_sorted[:n]] # Convert to 2D coordinates coords = np.array(np.unravel_index(sorted_idx, mean_map.shape)).T coords = coords + k // 2 mean_vals = [mean_map[i, j] for i, j in coords] results = list(zip(coords, mean_vals)) if print_coords: for (i, j), mean_val in results: print(f"Patch top-left ({i}, {j}) → mean={mean_val:.4f}") return results def _build_url(lat: float, lon: float) -> str: """Compose a Nominatim reverse-geocoding URL for the given coordinate.""" params = [ "format=jsonv2", f"lat={lat}", f"lon={lon}", "zoom=18", "addressdetails=1", "extratags=1", "namedetails=1", ] return f"{NOMINATIM_URL}?{'&'.join(params)}" def reverse_geocode(lat: float, lon: float) -> Dict[str, Any]: """Reverse-geocode a coordinate via Nominatim and return parsed address fields.""" url = _build_url(lat, lon) headers = {"User-Agent": os.getenv('USER_AGENT'), "Accept-Language": "en"} response = requests.request("GET", url, headers=headers, timeout=10) payload = response.json() address = payload.get("address") return { "country": address.get("country"), "state": address.get("state") or address.get("region"), "city": address.get("city") or address.get("town") or address.get("village"), "display_name": payload.get("display_name"), "raw": payload, } if __name__ == "__main__": r = reverse_geocode(lat = 40.703954037, lon = -73.86810149) print(r)