Oleksandr Kozachuk
112 lines
4.7 KiB
Python
112 lines
4.7 KiB
Python
"""Weather tool via MET Norway Locationforecast (SPEC-016).
|
|
|
|
A `get_weather` function tool: both personas talk about weather (the
|
|
sea over the skerries, rain on patch day) but had to guess it. The
|
|
free api.met.no compact forecast grounds it. Locations are
|
|
host-configured `[name, lat, lon]` entries — the model picks by name
|
|
and never supplies coordinates or URLs, so there is no SSRF surface.
|
|
"""
|
|
|
|
import logging
|
|
from typing import Any, Callable, Dict, List, Optional, Tuple
|
|
|
|
import aiohttp
|
|
|
|
from .ai_responder import sanitize_external_text
|
|
|
|
MET_COMPACT_URL = "https://api.met.no/weatherapi/locationforecast/2.0/compact"
|
|
USER_AGENT = "fjerkroa-discord-bot/3 (https://fjerkroa.no)"
|
|
FETCH_TIMEOUT_S = 15
|
|
FORECAST_POINT_INDICES = (6, 12, 24) # hourly series: ~6h/12h/24h ahead
|
|
|
|
GET_WEATHER_TOOL = {
|
|
"name": "get_weather",
|
|
"description": "Current weather and a short forecast for the configured local places. Use this whenever weather comes "
|
|
"up in conversation — never guess or invent weather. Returns current temperature (°C), wind (m/s) and conditions, "
|
|
"plus a few forecast points.",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"location": {"type": "string", "description": "Place name to look up; omit for the default (first configured) place."},
|
|
},
|
|
"required": [],
|
|
},
|
|
}
|
|
|
|
|
|
def _reduce(data: Any, name: str) -> Dict[str, Any]:
|
|
"""Compact MET timeseries -> {location, now, forecast[]} (WEA-02). Nothing else reaches the prompt."""
|
|
series = data.get("properties", {}).get("timeseries", []) if isinstance(data, dict) else []
|
|
if not series:
|
|
return {"error": "weather data unavailable"}
|
|
|
|
def point(entry: Dict[str, Any]) -> Dict[str, Any]:
|
|
details = entry.get("data", {}).get("instant", {}).get("details", {})
|
|
hour = entry.get("data", {}).get("next_1_hours", {}) or entry.get("data", {}).get("next_6_hours", {})
|
|
out: Dict[str, Any] = {
|
|
"time": str(entry.get("time", "")),
|
|
"temp_c": details.get("air_temperature"),
|
|
"wind_ms": details.get("wind_speed"),
|
|
}
|
|
symbol = hour.get("summary", {}).get("symbol_code")
|
|
if symbol:
|
|
out["conditions"] = str(symbol)
|
|
precip = hour.get("details", {}).get("precipitation_amount")
|
|
if precip is not None:
|
|
out["precip_mm"] = precip
|
|
return out
|
|
|
|
forecast = [point(series[i]) for i in FORECAST_POINT_INDICES if i < len(series)]
|
|
return {"location": sanitize_external_text(name, 80), "now": point(series[0]), "forecast": forecast}
|
|
|
|
|
|
class Weather:
|
|
def __init__(self, config_getter: Callable[[], Dict[str, Any]]) -> None:
|
|
self._config = config_getter
|
|
|
|
def _locations(self) -> List[Tuple[str, float, float]]:
|
|
out: List[Tuple[str, float, float]] = []
|
|
for entry in self._config().get("weather-locations", []):
|
|
try:
|
|
name, lat, lon = entry[0], float(entry[1]), float(entry[2])
|
|
out.append((str(name), lat, lon))
|
|
except (TypeError, ValueError, IndexError):
|
|
logging.warning(f"weather: bad location entry {entry!r}")
|
|
return out
|
|
|
|
def enabled(self) -> bool:
|
|
return bool(self._config().get("enable-weather", False)) and bool(self._locations())
|
|
|
|
def _pick(self, location: Optional[str]) -> Optional[Tuple[str, float, float]]:
|
|
"""Case-insensitive substring match; unknown/absent = first configured (WEA-03)."""
|
|
entries = self._locations()
|
|
if not entries:
|
|
return None
|
|
wanted = (location or "").strip().casefold()
|
|
if wanted:
|
|
for entry in entries:
|
|
if wanted in entry[0].casefold():
|
|
return entry
|
|
return entries[0]
|
|
|
|
async def _fetch_json(self, lat: float, lon: float) -> Any:
|
|
timeout = aiohttp.ClientTimeout(total=FETCH_TIMEOUT_S)
|
|
params = {"lat": f"{lat:.4f}", "lon": f"{lon:.4f}"}
|
|
async with aiohttp.ClientSession(timeout=timeout, headers={"User-Agent": USER_AGENT}) as session:
|
|
async with session.get(MET_COMPACT_URL, params=params) as response:
|
|
response.raise_for_status()
|
|
return await response.json()
|
|
|
|
async def forecast(self, location: Optional[str] = None) -> Dict[str, Any]:
|
|
"""Return a compact forecast, or an error dict — never raise (WEA-04)."""
|
|
picked = self._pick(location)
|
|
if picked is None:
|
|
return {"error": "weather unavailable: no locations configured"}
|
|
name, lat, lon = picked
|
|
try:
|
|
data = await self._fetch_json(lat, lon)
|
|
except Exception as err:
|
|
logging.warning(f"weather fetch failed: {err!r}")
|
|
return {"error": "weather lookup failed"}
|
|
return _reduce(data, name)
|