weather-base-station/environment/smhi.py

# environment/smhi.py

import logging
from datetime import datetime, timezone

import requests
from tools.geometry import Coordinate, LineString, MultiPolygon, Polygon

logger = logging.getLogger(__name__)


class SMHI:

    fire_warning_api = "https://opendata-download-metfcst.smhi.se/api/category/fwif1g/version/1/daily/geotype/point/lon/{lon}/lat/{lat}/data.json"
    forecast_api = "https://opendata-download-metfcst.smhi.se/api/category/snow1g/version/1/geotype/point/lon/{lon}/lat/{lat}/data.json"
    alerts_api = (
        "https://opendata-download-warnings.smhi.se/ibww/api/version/1/warning.json"
    )

    def __init__(self, location: Coordinate, session: requests.Session) -> None:
        self.location = location
        self._session = session
        self._data = {"fire_index": {}, "weather_alerts": {}, "forecast": {}}
        self._api = {
            "fire_index": self._get_fire_warning,
            "weather_alerts": self._get_alerts,
            "forecast": self._get_forecast,
        }

    def _calculate_area(self, area_type: str, area_list: list) -> bool:
        """Helper to check if our home coordinates fall within an SMHI warning zone."""
        point = self.location.to_Point(reversed=True)
        match area_type:
            case "Polygon":
                return Polygon(area_list[0]).contains(point)
            case "LineString":
                return LineString(area_list).distance_to(point) <= 0.045
            case "MultiPolygon":
                return MultiPolygon(area_list).contains(point)
        return False

    def _get_fire_warning(self) -> bool:
        try:
            resp = self._session.get(
                self.fire_warning_api.format(
                    lon=self.location.long, lat=self.location.lat
                ),
                timeout=10,
            )
            resp.raise_for_status()

            for param in resp.json()["timeSeries"][0]["parameters"]:
                if param["name"] == "fwiindex":
                    self._data["fire_index"] = {
                        "data": param["values"][0],
                        "valid_time": datetime.now(),
                    }
            return True
        except requests.RequestException as e:
            logger.error(f"Network error getting SMHI Fire index: {e}")
        except Exception as e:
            logger.error(f"Parsing error for SMHI Fire index: {e}")
        return False

    def _get_alerts(self) -> bool:
        try:
            resp = self._session.get(self.alerts_api, timeout=10)
            resp.raise_for_status()

            area_list = []
            now_utc = datetime.now(timezone.utc)

            for warning in resp.json():
                for areas in warning["warningAreas"]:
                    warning_level = areas["warningLevel"]["code"]

                    if warning_level == "MESSAGE":
                        continue

                    start_str = areas.get("approximateStart")
                    if start_str:
                        start_time = datetime.fromisoformat(
                            start_str.replace("Z", "+00:00")
                        )
                        if now_utc < start_time:
                            continue

                    end_str = areas.get("approximateEnd")
                    if end_str:
                        end_time = datetime.fromisoformat(
                            end_str.replace("Z", "+00:00")
                        )
                        if now_utc > end_time:
                            continue

                    is_in_area = False
                    if areas["area"]["type"] == "FeatureCollection":
                        for features in areas["area"]["features"]:
                            if self._calculate_area(
                                features["geometry"]["type"],
                                features["geometry"]["coordinates"],
                            ):
                                is_in_area = True
                                break
                    elif self._calculate_area(
                        areas["area"]["geometry"]["type"],
                        areas["area"]["geometry"]["coordinates"],
                    ):
                        is_in_area = True

                    if is_in_area:
                        area_list.append(
                            {
                                "areaName": areas["areaName"]["sv"],
                                "type": warning_level,
                                "description": areas["eventDescription"]["code"],
                            }
                        )

            self._data["weather_alerts"] = {
                "data": area_list,
                "valid_time": datetime.now(),
            }
            return True
        except requests.RequestException as e:
            logger.error(f"Network error getting SMHI Alerts: {e}")
        except Exception as e:
            logger.error(f"Parsing error for SMHI Alerts: {e}")
        return False

    def _get_forecast(self) -> bool:
        try:
            base_url = self.forecast_api.format(
                lon=self.location.long, lat=self.location.lat
            )

            query_string = "?timeseries=1&parameters=wind_from_direction,wind_speed,wind_speed_of_gust,symbol_code"
            full_url = base_url + query_string

            resp = self._session.get(
                full_url,
                timeout=10,
            )
            resp.raise_for_status()

            w_data = resp.json()["timeSeries"][0]["data"]

            self._data["forecast"] = {
                "data": {
                    "wind": {
                        "deg": w_data["wind_from_direction"],
                        "speed": round(w_data["wind_speed"], 1),
                        "gust": round(w_data["wind_speed_of_gust"], 1),
                    },
                    "weather_symbol": w_data["symbol_code"] - 1,
                },
                "valid_time": datetime.now(),
            }
            return True
        except requests.RequestException as e:
            logger.error(f"Network error getting SMHI Forecast: {e}")
        except KeyError as e:
            logger.error(f"Missing expected key in SMHI Forecast data: {e}")
        except Exception as e:
            logger.error(f"Parsing error for SMHI Forecast: {e}")
        return False

    def get(self) -> dict:
        """Returns valid SMHI data, utilizing cached data if recently fetched."""
        valid_data = {}
        for api_name, fetch_func in self._api.items():
            cache = self._data[api_name]
            valid_time = cache.get("valid_time", datetime.fromtimestamp(0))
            age_seconds = (datetime.now() - valid_time).total_seconds()

            if age_seconds <= 300:
                logger.debug(f'Using cached data for "{api_name}"')
                valid_data[api_name] = cache["data"]
            else:
                logger.debug(f'Fetching new data for "{api_name}"...')
                if fetch_func():
                    logger.debug(f'Successfully fetched new data for "{api_name}"')
                    valid_data[api_name] = self._data[api_name]["data"]
                elif age_seconds <= 600:
                    logger.warning(
                        f'Failed fetching "{api_name}", falling back to 10-min cache.'
                    )
                    valid_data[api_name] = cache["data"]
                else:
                    logger.error(
                        f'SMHI data "{api_name}" is too old and cannot be refreshed.'
                    )

        return valid_data