diff --git a/tests/unit/test_weather.py b/tests/unit/test_weather.py new file mode 100644 index 0000000..f55722a --- /dev/null +++ b/tests/unit/test_weather.py @@ -0,0 +1,39 @@ +import time +from unittest.mock import patch, MagicMock +from vigilar.detection.weather import WeatherFetcher, _weather_code_to_text + + +def test_get_conditions_returns_dict_on_success(): + fetcher = WeatherFetcher() + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = {"current": {"temperature_2m": 15.5, "weather_code": 2}} + with patch("vigilar.detection.weather.requests.get", return_value=mock_response): + result = fetcher.get_conditions(45.0, -85.0) + assert result is not None + assert result["temperature_c"] == 15.5 + assert isinstance(result["conditions"], str) + + +def test_get_conditions_returns_none_on_failure(): + fetcher = WeatherFetcher() + with patch("vigilar.detection.weather.requests.get", side_effect=Exception("offline")): + result = fetcher.get_conditions(45.0, -85.0) + assert result is None + + +def test_get_conditions_caches(): + fetcher = WeatherFetcher() + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = {"current": {"temperature_2m": 10.0, "weather_code": 0}} + with patch("vigilar.detection.weather.requests.get", return_value=mock_response) as mock_get: + fetcher.get_conditions(45.0, -85.0) + fetcher.get_conditions(45.0, -85.0) + assert mock_get.call_count == 1 + + +def test_weather_code_mapping(): + assert _weather_code_to_text(0) == "Clear sky" + assert _weather_code_to_text(61) == "Light rain" + assert _weather_code_to_text(999) == "Unknown" diff --git a/vigilar/detection/weather.py b/vigilar/detection/weather.py new file mode 100644 index 0000000..0bd1bad --- /dev/null +++ b/vigilar/detection/weather.py @@ -0,0 +1,57 @@ +"""Open-Meteo weather fetcher with in-memory caching.""" + +import logging +import time + +import requests + +log = logging.getLogger(__name__) + +_WMO_CODES = { + 0: "Clear sky", 1: "Mainly clear", 2: "Partly cloudy", 3: "Overcast", + 45: "Fog", 48: "Depositing rime fog", + 51: "Light drizzle", 53: "Moderate drizzle", 55: "Dense drizzle", + 61: "Light rain", 63: "Moderate rain", 65: "Heavy rain", + 66: "Light freezing rain", 67: "Heavy freezing rain", + 71: "Light snow", 73: "Moderate snow", 75: "Heavy snow", 77: "Snow grains", + 80: "Light showers", 81: "Moderate showers", 82: "Violent showers", + 85: "Light snow showers", 86: "Heavy snow showers", + 95: "Thunderstorm", 96: "Thunderstorm with light hail", 99: "Thunderstorm with heavy hail", +} + +CACHE_TTL_S = 3600 + + +def _weather_code_to_text(code: int) -> str: + return _WMO_CODES.get(code, "Unknown") + + +class WeatherFetcher: + def __init__(self): + self._cache: dict[str, tuple[dict, float]] = {} + + def get_conditions(self, lat: float, lon: float) -> dict | None: + cache_key = f"{lat:.2f},{lon:.2f}" + if cache_key in self._cache: + data, ts = self._cache[cache_key] + if time.time() - ts < CACHE_TTL_S: + return data + try: + resp = requests.get( + "https://api.open-meteo.com/v1/forecast", + params={"latitude": lat, "longitude": lon, "current": "temperature_2m,weather_code"}, + timeout=10, + ) + if resp.status_code != 200: + return None + j = resp.json() + current = j.get("current", {}) + result = { + "temperature_c": current.get("temperature_2m"), + "conditions": _weather_code_to_text(current.get("weather_code", -1)), + } + self._cache[cache_key] = (result, time.time()) + return result + except Exception: + log.debug("Weather fetch failed", exc_info=True) + return None