smarter: factual-model routing (beh-10), fetch_url main-content extraction + 8k cap (url-08), get_weather via met.no (spec-016)

fjerkroa_bot/weather.py

@@ -0,0 +1,111 @@
+"""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)