"""Small replaceable weather-provider boundary and Open-Meteo implementation.""" from __future__ import annotations import hashlib import json import math from dataclasses import asdict, dataclass, field from datetime import datetime, timezone from pathlib import Path from typing import Any, Protocol from urllib.parse import urlencode from zoneinfo import ZoneInfo from mountain_twin.analysis_contract import WeatherSourceType from mountain_twin.offline import offline_mode, urlopen # AV-045: offline mode for tests from mountain_twin.snow.contracts import ( MODELLED_GROUND_SNOW_DEPTH_SEMANTICS_V0_1, MODELLED_SNOWFALL_SEMANTICS_V0_1, ) HOURLY_VARIABLES = ( "temperature_2m", "relative_humidity_2m", "apparent_temperature", "precipitation", "rain", "snowfall", "snow_depth", "precipitation_probability", "cloud_cover", "cloud_cover_low", "cloud_cover_mid", "cloud_cover_high", "wind_speed_10m", "wind_direction_10m", "wind_gusts_10m", "visibility", "freezing_level_height", "weather_code", ) FORECAST_ENDPOINT = "https://api.open-meteo.com/v1/forecast" HISTORICAL_FORECAST_ENDPOINT = "https://historical-forecast-api.open-meteo.com/v1/forecast" @dataclass(frozen=True) class WeatherLocation: sample_id: str point_index: int route_distance_m: float latitude: float longitude: float route_elevation_m: float | None selection_reasons: tuple[str, ...] @dataclass(frozen=True) class WeatherRequest: locations: tuple[WeatherLocation, ...] scenario_datetime: datetime source_type: WeatherSourceType timezone: str model_selection: str = "best_match" hourly_variables: tuple[str, ...] = HOURLY_VARIABLES def __post_init__(self) -> None: if not self.locations: raise ValueError("weather request requires at least one location") if self.scenario_datetime.utcoffset() is None: raise ValueError("weather scenario datetime must be timezone-aware") if self.source_type not in { WeatherSourceType.FORECAST, WeatherSourceType.HISTORICAL_FORECAST, }: raise ValueError("Open-Meteo v0.1 supports forecast source types only") @dataclass(frozen=True) class WeatherSample: location: WeatherLocation source_type: WeatherSourceType provider: str api_family: str endpoint_type: str model_selection: str model_identity: str | None provider_valid_time: str provider_run_time: str | None provider_grid_latitude: float | None provider_grid_longitude: float | None provider_grid_elevation_m: float | None variables: dict[str, float | int | None] units: dict[str, str | None] missing_variable_reasons: dict[str, str] variable_semantics: dict[str, dict[str, Any]] = field(default_factory=dict) @property def route_minus_provider_elevation_m(self) -> float | None: if self.location.route_elevation_m is None or self.provider_grid_elevation_m is None: return None return self.location.route_elevation_m - self.provider_grid_elevation_m class WeatherProvider(Protocol): def fetch(self, request: WeatherRequest) -> tuple[WeatherSample, ...]: """Return values for the explicit scenario time at every requested location.""" class OpenMeteoProvider: """Open-Meteo R&D adapter; raw responses are cached locally, never committed.""" provider_name = "Open-Meteo" def __init__(self, cache_directory: Path, *, opener=urlopen) -> None: self.cache_directory = cache_directory self.opener = opener def cache_key(self, request: WeatherRequest) -> str: payload = { "endpoint": self._endpoint(request.source_type), "locations": [asdict(location) for location in request.locations], "scenario_datetime": request.scenario_datetime.isoformat(), "source_type": request.source_type.value, "timezone": request.timezone, "model_selection": request.model_selection, "hourly_variables": list(request.hourly_variables), } return hashlib.sha256( json.dumps(payload, sort_keys=True, separators=(",", ":")).encode() ).hexdigest() def fetch(self, request: WeatherRequest) -> tuple[WeatherSample, ...]: key = self.cache_key(request) cache_path = self.cache_directory / f"{key}.json" metadata_path = self.cache_directory / f"{key}.metadata.json" if cache_path.exists(): raw = json.loads(cache_path.read_text(encoding="utf-8")) else: url = self._url(request) try: with self.opener(url, timeout=30) as response: raw = json.loads(response.read().decode("utf-8")) except OSError as error: raise RuntimeError("WEATHER_PROVIDER_NETWORK_FAILURE") from error if self.opener is urlopen and offline_mode() is not None: # AV-045: an offline-mode answer never lands in the disk cache. return self._parse(request, raw) self.cache_directory.mkdir(parents=True, exist_ok=True) cache_path.write_text( json.dumps(raw, sort_keys=True, indent=2) + "\n", encoding="utf-8" ) metadata_path.write_text( json.dumps( {"fetched_at": datetime.now(timezone.utc).isoformat()}, sort_keys=True, indent=2, ) + "\n", encoding="utf-8", ) return self._parse(request, raw) def request_provenance(self, request: WeatherRequest) -> dict[str, Any]: """Return portable provenance for a cached or live request.""" key = self.cache_key(request) metadata_path = self.cache_directory / f"{key}.metadata.json" metadata = ( json.loads(metadata_path.read_text(encoding="utf-8")) if metadata_path.exists() else {} ) return { "endpoint": self._endpoint(request.source_type), "endpoint_type": "forecast" if request.source_type is WeatherSourceType.FORECAST else "historical_forecast", "cache_key": key, "fetched_at": metadata.get("fetched_at"), "model_selection": request.model_selection, "query_coordinates": [ {"latitude": item.latitude, "longitude": item.longitude} for item in request.locations ], "hourly_variables": list(request.hourly_variables), "timezone": request.timezone, } def _url(self, request: WeatherRequest) -> str: scenario = request.scenario_datetime.astimezone(ZoneInfo(request.timezone)) params = { "latitude": ",".join(str(location.latitude) for location in request.locations), "longitude": ",".join(str(location.longitude) for location in request.locations), "hourly": ",".join(request.hourly_variables), "timezone": request.timezone, "models": request.model_selection, "start_date": scenario.date().isoformat(), "end_date": scenario.date().isoformat(), } return f"{self._endpoint(request.source_type)}?{urlencode(params)}" @staticmethod def _endpoint(source_type: WeatherSourceType) -> str: if source_type is WeatherSourceType.FORECAST: return FORECAST_ENDPOINT if source_type is WeatherSourceType.HISTORICAL_FORECAST: return HISTORICAL_FORECAST_ENDPOINT raise ValueError(f"Open-Meteo v0.1 does not implement {source_type.value}") def _parse(self, request: WeatherRequest, raw: Any) -> tuple[WeatherSample, ...]: responses = raw if isinstance(raw, list) else [raw] if len(responses) != len(request.locations): raise ValueError("WEATHER_PROVIDER_PARTIAL_LOCATION_RESPONSE") expected_time = request.scenario_datetime.astimezone(ZoneInfo(request.timezone)).strftime( "%Y-%m-%dT%H:00" ) result = [] for location, response in zip(request.locations, responses): if "error" in response: raise ValueError(f"WEATHER_PROVIDER_ERROR: {response.get('reason', 'unknown')}") hourly = response.get("hourly", {}) times = hourly.get("time", []) if expected_time not in times: raise ValueError("WEATHER_PROVIDER_VALID_TIME_MISSING") index = times.index(expected_time) units = response.get("hourly_units", {}) values, missing, variable_semantics = {}, {}, {} for variable in request.hourly_variables: semantics = weather_variable_semantics(variable) if semantics: variable_semantics[variable] = semantics series = hourly.get(variable) if series is None: values[variable] = None missing[variable] = "PROVIDER_VARIABLE_UNSUPPORTED_OR_ABSENT" elif index >= len(series) or series[index] is None: values[variable] = None missing[variable] = "PROVIDER_VALUE_MISSING_AT_VALID_TIME" else: invalid_reason = _invalid_value_reason( variable, series[index], units.get(variable) ) if invalid_reason is not None: values[variable] = None missing[variable] = invalid_reason else: values[variable] = series[index] result.append( WeatherSample( location=location, source_type=request.source_type, provider=self.provider_name, api_family="Open-Meteo Weather API", endpoint_type="forecast" if request.source_type is WeatherSourceType.FORECAST else "historical_forecast", model_selection=request.model_selection, model_identity=response.get("model") or response.get("model_name"), provider_valid_time=expected_time, provider_run_time=_run_time(response), provider_grid_latitude=response.get("latitude"), provider_grid_longitude=response.get("longitude"), provider_grid_elevation_m=response.get("elevation"), variables=values, units={variable: units.get(variable) for variable in request.hourly_variables}, missing_variable_reasons=missing, variable_semantics=variable_semantics, ) ) return tuple(result) def weather_variable_semantics(variable: str) -> dict[str, Any]: """Return explicit Snow semantics for provider fields that have them.""" semantics = { "snowfall": MODELLED_SNOWFALL_SEMANTICS_V0_1, "snow_depth": MODELLED_GROUND_SNOW_DEPTH_SEMANTICS_V0_1, }.get(variable) return semantics.to_dict() if semantics is not None else {} def _invalid_value_reason(variable: str, value: Any, unit: str | None) -> str | None: """Reject invalid modelled depth values without silently correcting them.""" if variable != "snow_depth": return None if unit != "m": return "PROVIDER_VARIABLE_UNIT_UNSUPPORTED" if ( isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value) or value < 0 ): return "PROVIDER_VALUE_INVALID_AT_VALID_TIME" return None def _run_time(response: dict[str, Any]) -> str | None: value = response.get("run") or response.get("run_time") or response.get("model_run") if value is None: return None if isinstance(value, (int, float)): return datetime.fromtimestamp(value, timezone.utc).isoformat() return str(value)