Spaces:
Sleeping
Sleeping
File size: 3,132 Bytes
93c158c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 | 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() |