from __future__ import annotations import os from typing import Dict, Any, Optional, Tuple, List import requests from dateutil import tz from retrying import retry from .utils import get_env class GeoCoder: """Simple geocoding via Open-Meteo Nominatim proxy (no key).""" @staticmethod def geocode(place: str) -> Optional[Tuple[float, float, str]]: try: url = f"https://geocoding-api.open-meteo.com/v1/search?name={requests.utils.quote(place)}&count=1&language=en&format=json" r = requests.get(url, timeout=15) r.raise_for_status() data = r.json() if data.get("results"): item = data["results"][0] return float(item["latitude"]), float(item["longitude"]), item.get("name") + ", " + item.get("country", "") except Exception: return None return None class OpenMeteoProvider: """Free fallback provider for forecast and climate normals.""" @staticmethod def forecast(lat: float, lon: float) -> Dict[str, Any]: url = ( "https://api.open-meteo.com/v1/forecast?" f"latitude={lat}&longitude={lon}&hourly=temperature_2m,relative_humidity_2m,precipitation,wind_speed_10m" "&daily=temperature_2m_max,temperature_2m_min,precipitation_sum,wind_speed_10m_max&forecast_days=14&timezone=auto" ) r = requests.get(url, timeout=20) r.raise_for_status() return r.json() @staticmethod def climate_normals(lat: float, lon: float) -> Dict[str, Any]: # Monthly normals & warming continuation heuristic inputs url = ( "https://climate-api.open-meteo.com/v1/climate?" f"latitude={lat}&longitude={lon}&models=ERA5&month=1..12&daily=temperature_2m_mean,precipitation_sum" ) r = requests.get(url, timeout=25) r.raise_for_status() return r.json() class WeatherAPIProvider: base = "http://api.weatherapi.com/v1" @staticmethod def enabled() -> bool: return bool(get_env("WEATHERAPI_KEY")) @staticmethod def forecast(q: str, days: int = 14) -> Optional[Dict[str, Any]]: if not WeatherAPIProvider.enabled(): return None url = f"{WeatherAPIProvider.base}/forecast.json?key={get_env('WEATHERAPI_KEY')}&q={requests.utils.quote(q)}&days={min(days,14)}&aqi=yes&alerts=yes" r = requests.get(url, timeout=20) if r.status_code != 200: return None return r.json() class OpenWeatherProvider: base = "https://api.openweathermap.org/data/2.5/onecall" @staticmethod def enabled() -> bool: return bool(get_env("OPENWEATHER_API_KEY")) @staticmethod def onecall(lat: float, lon: float) -> Optional[Dict[str, Any]]: if not OpenWeatherProvider.enabled(): return None url = f"{OpenWeatherProvider.base}?lat={lat}&lon={lon}&exclude=minutely&appid={get_env('OPENWEATHER_API_KEY')}&units=metric" r = requests.get(url, timeout=20) if r.status_code != 200: return None return r.json()