"""Live Open-Meteo lookup for one arbitrary point (a camp the frozen TMB/ Tatra weather fixtures do not cover), with a simple in-process cache so the same point/night is fetched once, not once per render. Reuses mountain_twin.weather.provider.OpenMeteoProvider's HTTP endpoint selection, hourly-variable list, unit/semantics helpers and on-disk raw- response cache directory. It does not call OpenMeteoProvider.fetch() directly: that method's contract deliberately extracts a single instant's values (mountain_twin/weather/provider.py's _parse(), "expected_time"), which fits the existing per-scenario Unified Route Analysis use case but discards the rest of the day's hourly array. An overnight window needs the whole series (and often spans two calendar dates), so this module performs its own minimal fetch + parse into the same mountain_twin.weather.temporal. TemporalWeatherSample shape temporal_window_value() already consumes -- every other piece (endpoint, variables, validity checks) is the existing, untouched provider code. Both cache layers (in-memory and the raw responses on disk) expire after LIVE_WEATHER_TTL_SECONDS: a forecast is a moving target, so after the TTL the same point is fetched fresh instead of reusing an outdated run. """ from __future__ import annotations import hashlib import json import re import threading import time from dataclasses import replace from datetime import date, datetime, timedelta, timezone from pathlib import Path from typing import Any from urllib.error import HTTPError from urllib.parse import urlencode from zoneinfo import ZoneInfo from mountain_twin import usage as service_usage from mountain_twin.analysis_contract import WeatherSourceType from mountain_twin.budgets import ServiceBudgetExceeded from mountain_twin.offline import ( # AV-045: offline mode for tests offline_cache_directory, offline_mode, urlopen, ) from mountain_twin.weather.forecast_cache import ( MAX_STALE_SECONDS, ForecastCellCache, call_source, call_weight, cell_key, current_call_source, effective_ttl, grid_cell, ) from mountain_twin.weather.provider import ( HOURLY_VARIABLES, OpenMeteoProvider, _invalid_value_reason, weather_variable_semantics, ) from mountain_twin.weather.spatial import RouteWeatherSample from mountain_twin.weather.temporal import TemporalWeatherSample # One hour. Open-Meteo's own model metadata (checked 2026-09-27) shows the # models behind "best_match" publishing new runs every 1 h (MET Nordic), 3 h # (AROME, ICON-D2/EU -- the Alps) and 6 h (ICON, ECMWF, GFS), with runs of # different models landing at staggered times through each hour. A 1 h TTL # keeps a cached forecast at most about one fresh run behind; 3 h could miss # a whole regional cycle. Cost: at most one request per point per hour. LIVE_WEATHER_TTL_SECONDS = 3600 # Stored inside each on-disk raw response: when it was fetched (epoch s). # A cached file without it (written before the TTL existed) counts as expired. FETCHED_AT_KEY = "_aventurro_fetched_at" # Open-Meteo answers a burst above its per-minute limit with 429; such a # request is retried (see LiveWeatherPointResolver._fetch_json). RATE_LIMIT_ATTEMPTS = 3 RATE_LIMIT_MAX_WAIT_SECONDS = 5.0 # A 429's own reason says which limit was hit. The per-minute one is a burst # (retried above); the hourly and daily ones mean the quota is spent -- # retrying cannot help until it renews (the daily one: "try again tomorrow"), # so those fail at once. Failures are never cached: the next request asks # again (docs: 429 is not a network failure and is reported as what it is). RATE_LIMIT_REASON_BY_PERIOD = { "MINUTE": "PROVIDER_RATE_LIMITED", "HOURLY": "PROVIDER_HOURLY_LIMIT", "DAILY": "PROVIDER_DAILY_LIMIT", } class ProviderRateLimited(RuntimeError): """Open-Meteo answered 429. ``period`` is MINUTE, HOURLY or DAILY; ``reason_code`` the matching unavailable reason; ``provider_reason`` the provider's own words.""" def __init__(self, period: str, provider_reason: str | None): super().__init__("WEATHER_PROVIDER_RATE_LIMITED") self.period = period self.reason_code = RATE_LIMIT_REASON_BY_PERIOD[period] self.provider_reason = provider_reason def rate_limit_until(period: str, at_epoch: float) -> float: """AV-051: when a limit's quota comes back -- the next minute, the next full hour (UTC), the next UTC midnight (Open-Meteo counts its days in UTC) -- for the page to say "wróci ok. HH:MM", never "daily" for an hourly limit.""" moment = datetime.fromtimestamp(at_epoch, timezone.utc) if period == "MINUTE": return at_epoch + 60 if period == "HOURLY": return (moment.replace(minute=0, second=0, microsecond=0) + timedelta(hours=1)).timestamp() return ( moment.replace(hour=0, minute=0, second=0, microsecond=0) + timedelta(days=1) ).timestamp() def _rate_limit_period(error: HTTPError) -> tuple[str, str | None]: try: reason = json.loads(error.read().decode("utf-8")).get("reason") except (OSError, ValueError, AttributeError): reason = None text = (reason or "").lower() period = "DAILY" if "daily" in text else "HOURLY" if "hourly" in text else "MINUTE" return period, reason # AV-051: a cell's answer covers at most two weeks unless more is asked for. FETCH_MAX_DAYS = 14 # AV-051: MET Norway answers what Open-Meteo could not -- these failures -- # or everything once the day's Open-Meteo count reaches this share of its # limit; one request per grid cell, at most this many per lookup. FALLBACK_FAILURES = frozenset( { "PROVIDER_DAILY_LIMIT", "PROVIDER_HOURLY_LIMIT", "PROVIDER_RATE_LIMITED", "PROVIDER_NETWORK_FAILURE", "OPEN_METEO_NOT_ASKED", # AV-052: our own RED and STOP levels leave the weather to MET Norway. "SERVICE_BUDGET_RED", "SERVICE_BUDGET_STOP", } ) MET_FALLBACK_THRESHOLD = 0.9 MET_FALLBACK_MAX_LOCATIONS = 150 # Open-Meteo's multi-location requests take up to 100 coordinates. BATCH_MAX_LOCATIONS = 100 # Freezing level has regional gaps under the default "best_match" selection. # Live queries (2026-09-27, 72 h, docs/design_reference/ # live_route_weather_spike_v0_1.md section 6): best_match returns 0/72 values # over mainland Norway (Lofoten, Jotunheimen, Tromso), where it resolves to # MET Nordic, which publishes no freezing level, nor do ECMWF IFS, GEM, UKMO, # Meteo-France, JMA, KNMI. ICON (icon_seamless: ICON-D2/EU/global) and GFS # return 72/72 there and everywhere else tested (Alps, Tatras, Svalbard, # Andes). So when the default series has no value at all, the freezing level # alone is fetched from ICON and labelled with that model; if ICON has none # either, it is reported as unavailable for the region -- never 0, never an # unlabelled substitute. FREEZING_LEVEL_VARIABLE = "freezing_level_height" FREEZING_LEVEL_FALLBACK_MODEL = "icon_seamless" FREEZING_LEVEL_NOT_PROVIDED_FOR_REGION = "NOT_PROVIDED_FOR_REGION" FREEZING_LEVEL_FALLBACK_FETCH_FAILED = "FALLBACK_MODEL_FETCH_FAILED" # ~11 m at the equator: enough to treat two renders of "the same camp pin" # as the same point without pretending sub-metre GPS precision. COORDINATE_CACHE_PRECISION = 4 ELEVATION_CACHE_BUCKET_M = 10.0 # The variables this resolver requests: the provider's list plus what the # Mission Control cockpit adds (docs/cockpit_v0_1_design.md section 2) -- # kept separate so the Unified Route Analysis request and its cache keys do # not change. soil_temperature_0cm is the real field name (the API rejects # ground_temperature / soil_temperature). # Storms tab (docs/profile_storms_v0_1_design.md section 4): the model's # convective variables and weather code, fetched in the same request as the # rest (one cache entry serves the Weather and the Storms tab). STORM_HOURLY_VARIABLES: tuple[str, ...] = ( "cape", "lifted_index", "convective_inhibition", "weather_code", ) LIVE_HOURLY_VARIABLES: tuple[str, ...] = ( *HOURLY_VARIABLES, "soil_temperature_0cm", *(variable for variable in STORM_HOURLY_VARIABLES if variable not in HOURLY_VARIABLES), ) def _bucketed_elevation(elevation_m: float | None) -> float | None: if elevation_m is None: return None return round(elevation_m / ELEVATION_CACHE_BUCKET_M) * ELEVATION_CACHE_BUCKET_M def _cache_key( latitude: float, longitude: float, elevation_m: float | None, start: date, end: date, timezone_name: str, ): bucketed_elevation = ( round((elevation_m or 0.0) / ELEVATION_CACHE_BUCKET_M) * ELEVATION_CACHE_BUCKET_M ) return ( round(latitude, COORDINATE_CACHE_PRECISION), round(longitude, COORDINATE_CACHE_PRECISION), bucketed_elevation, start.isoformat(), end.isoformat(), timezone_name, ) def _keep_on_disk(path: Path, raw: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(raw, sort_keys=True, indent=2) + "\n", encoding="utf-8") class LiveWeatherPointResolver: """TTL cache (memory + disk) in front of a live Open-Meteo hourly-series fetch. ``clock`` returns epoch seconds; injectable for tests.""" def __init__( self, provider: OpenMeteoProvider, *, opener=urlopen, clock=time.time, ttl_seconds: float = LIVE_WEATHER_TTL_SECONDS, sleep=time.sleep, cell_cache: ForecastCellCache | None = None, refresher=None, log=None, fallback=None, primary: str = "open_meteo", ) -> None: """AV-048: with ``cell_cache`` the multi-point lookups (resolve_many, resolve_many_models) go through the shared grid-cell cache instead of per-point files; with ``refresher`` (a callable that runs a task later, e.g. an executor's submit) a stale answer is served at once and fetched again in the background; ``log`` gets one line per provider request with the day's call count.""" self._provider = provider self._opener = opener self._clock = clock self._sleep = sleep self._ttl_seconds = ttl_seconds self._cells = cell_cache self._refresher = refresher self._log = log # AV-051: MET Norway (mountain_twin.weather.met_norway.MetNorwayClient) # when Open-Meteo refuses or nears its limit; ``primary`` met_norway # asks it first (WEATHER_PRIMARY). self._fallback = fallback self._primary = primary if primary in ("open_meteo", "met_norway") else "open_meteo" self._in_flight: set[str] = set() self._in_flight_lock = threading.Lock() # key -> (fetched_at epoch seconds, sample or None) self._memory_cache: dict[tuple, tuple[float, TemporalWeatherSample | None]] = {} self.live_fetch_count = 0 # test/observability hook: real network calls only def _is_fresh(self, fetched_at: float | None) -> bool: return fetched_at is not None and self._clock() - fetched_at < self._ttl_seconds def resolve( self, *, latitude: float, longitude: float, elevation_m: float | None, start_date: date, end_date: date, timezone_name: str, source_type: WeatherSourceType = WeatherSourceType.FORECAST, ) -> TemporalWeatherSample | None: """Return the real hourly series for [start_date, end_date], or None if the provider has no usable data for this point (never a guess).""" key = _cache_key(latitude, longitude, elevation_m, start_date, end_date, timezone_name) cached = self._memory_cache.get(key) if cached is not None and self._is_fresh(cached[0]): return cached[1] fetched_at, sample = self._resolve_uncached( latitude, longitude, elevation_m, start_date, end_date, timezone_name, source_type ) self._memory_cache[key] = (fetched_at, sample) return sample def _resolve_uncached( self, latitude, longitude, elevation_m, start_date, end_date, timezone_name, source_type ): fetched_at, raw = self._raw_response( latitude, longitude, elevation_m, start_date, end_date, timezone_name, source_type ) if raw is None: return fetched_at, None sample = _parse_hourly_series(raw, latitude, longitude, elevation_m, timezone_name) if sample is None or _has_any_value(sample.variables_by_time.get(FREEZING_LEVEL_VARIABLE)): return fetched_at, sample try: fallback_fetched_at, fallback_raw = self._raw_response( latitude, longitude, elevation_m, start_date, end_date, timezone_name, source_type, model=FREEZING_LEVEL_FALLBACK_MODEL, variables=(FREEZING_LEVEL_VARIABLE,), ) except RuntimeError: return fetched_at, _without_freezing_level(sample, FREEZING_LEVEL_FALLBACK_FETCH_FAILED) # The combined entry is only as fresh as its older half. fetched_at = min(fetched_at, fallback_fetched_at) series = _aligned_fallback_series(sample, fallback_raw, timezone_name) if not _has_any_value(series): return fetched_at, _without_freezing_level( sample, FREEZING_LEVEL_NOT_PROVIDED_FOR_REGION ) # The unit comes with the values: where best_match has no freezing # level, Open-Meteo reports its unit as the string "undefined". fallback_unit = (fallback_raw or {}).get("hourly_units", {}).get(FREEZING_LEVEL_VARIABLE) return fetched_at, replace( sample, base=replace( sample.base, units={**sample.base.units, FREEZING_LEVEL_VARIABLE: fallback_unit}, ), variables_by_time={**sample.variables_by_time, FREEZING_LEVEL_VARIABLE: series}, variable_models={ **sample.variable_models, FREEZING_LEVEL_VARIABLE: FREEZING_LEVEL_FALLBACK_MODEL, }, ) def _raw_response( self, latitude, longitude, elevation_m, start_date, end_date, timezone_name, source_type, *, model: str | None = None, variables: tuple[str, ...] = LIVE_HOURLY_VARIABLES, ) -> tuple[float, dict[str, Any] | None]: """(fetched_at, raw) -- a fresh disk copy when there is one, otherwise a live fetch; the fetch time travels with the data so the memory entry expires when the disk copy it came from does.""" cache_path = self._cache_path( self._payload_key( latitude, longitude, elevation_m, start_date, end_date, timezone_name, source_type, model, variables, ) ) if cache_path.exists(): cached = json.loads(cache_path.read_text(encoding="utf-8")) if self._is_fresh(cached.get(FETCHED_AT_KEY)): return cached[FETCHED_AT_KEY], cached url = self._build_url( latitude, longitude, elevation_m, start_date, end_date, timezone_name, source_type, model, variables, ) raw = self._fetch_json(url) fetched_at = self._clock() if "error" in raw: return fetched_at, None raw[FETCHED_AT_KEY] = fetched_at _keep_on_disk(cache_path, raw) return fetched_at, raw def _fetch_json(self, url: str) -> dict[str, Any]: """One live request; "429 Too Many Requests" for Open-Meteo's per-minute limit (a burst such as a cockpit's first load) is retried after the server's Retry-After, or 1 s then 2 s, up to RATE_LIMIT_ATTEMPTS tries, then raised as ProviderRateLimited; a 429 for the hourly or daily quota is raised at once (retrying cannot help). Any other failure is a network failure at once.""" for attempt in range(1, RATE_LIMIT_ATTEMPTS + 1): self.live_fetch_count += 1 try: with self._opener(url, timeout=30) as response: return json.loads(response.read().decode("utf-8")) except HTTPError as error: if error.code != 429: raise RuntimeError("WEATHER_PROVIDER_NETWORK_FAILURE") from error period, reason = _rate_limit_period(error) if period != "MINUTE" or attempt == RATE_LIMIT_ATTEMPTS: raise ProviderRateLimited(period, reason) from error retry_after = error.headers.get("Retry-After") if error.headers else None try: delay = min(float(retry_after), RATE_LIMIT_MAX_WAIT_SECONDS) except (TypeError, ValueError): delay = float(attempt) self._sleep(delay) except OSError as error: raise RuntimeError("WEATHER_PROVIDER_NETWORK_FAILURE") from error raise AssertionError("unreachable") def resolve_many( self, points: list[tuple[float, float, float | None]], *, start_date: date, end_date: date, timezone_name: str, variables: tuple[str, ...], source_type: WeatherSourceType = WeatherSourceType.FORECAST, freshness: dict | None = None, ) -> list[tuple[TemporalWeatherSample | None, str | None]]: """Hourly series for many (latitude, longitude, elevation) points at once: each point is served from the same per-point TTL cache (memory and disk) as ``resolve``, and the rest is fetched in multi-location requests of up to BATCH_MAX_LOCATIONS -- the provider counts its quota per location, so batching saves requests and time, not quota (docs/design_reference/pogoda_spike_v0_1.md section a). Returns (sample, failure_reason) per point: a failed request is that point's PROVIDER_NETWORK_FAILURE; (None, None) is no usable provider data. AV-048: with the shared cell cache, ``freshness`` (a dict, optional) gets when the oldest answer used was fetched and whether stale ones are being refreshed in the background.""" if self._cells is not None: met_first = self._fallback is not None and ( self._primary == "met_norway" or self._near_limit() ) answers = ( [(None, "OPEN_METEO_NOT_ASKED")] * len(points) if met_first else self._cell_answers( points, model=None, variables=variables, start_date=start_date, end_date=end_date, timezone_name=timezone_name, source_type=source_type, freshness=freshness, ) ) samples = [ _parse_hourly_series(raw, *point, timezone_name, variables) if raw else None for point, (raw, _) in zip(points, answers) ] if not met_first: samples = self._with_freezing_fallback_cells( samples, points, start_date, end_date, timezone_name, source_type, freshness ) results = [(sample, failure) for sample, (_, failure) in zip(samples, answers)] if self._fallback is not None: results = self._met_norway_for_failures( results, points, start_date, end_date, timezone_name, variables ) return results results: list[tuple[TemporalWeatherSample | None, str | None]] = [(None, None)] * len( points ) keys = [ _cache_key(lat, lon, ele, start_date, end_date, timezone_name) + (variables,) for lat, lon, ele in points ] pending: list[int] = [] for position, (point, key) in enumerate(zip(points, keys)): cached = self._memory_cache.get(key) if cached is not None and self._is_fresh(cached[0]): results[position] = (cached[1], None) continue path = self._cache_path( self._payload_key( *point, start_date, end_date, timezone_name, source_type, None, variables ) ) if path.exists(): raw = json.loads(path.read_text(encoding="utf-8")) if self._is_fresh(raw.get(FETCHED_AT_KEY)): sample = self._with_freezing_fallback_many( [_parse_hourly_series(raw, *point, timezone_name, variables)], [point], start_date, end_date, timezone_name, source_type, )[0] self._memory_cache[key] = (raw[FETCHED_AT_KEY], sample) results[position] = (sample, None) continue pending.append(position) for first in range(0, len(pending), BATCH_MAX_LOCATIONS): chunk = pending[first : first + BATCH_MAX_LOCATIONS] chunk_points = [points[position] for position in chunk] try: raws = self._fetch_batch( chunk_points, start_date, end_date, timezone_name, source_type, None, variables ) except ProviderRateLimited as limited: # The hourly/daily quota is spent: the remaining batches would # get the same answer -- do not send them. remaining = chunk if limited.period == "MINUTE" else pending[first:] for position in remaining: results[position] = (None, limited.reason_code) if limited.period == "MINUTE": continue break except RuntimeError: for position in chunk: results[position] = (None, "PROVIDER_NETWORK_FAILURE") continue fetched_at = self._clock() samples = [] for point, raw in zip(chunk_points, raws): if raw is None or "error" in raw: samples.append(None) continue raw[FETCHED_AT_KEY] = fetched_at path = self._cache_path( self._payload_key( *point, start_date, end_date, timezone_name, source_type, None, variables ) ) _keep_on_disk(path, raw) samples.append(_parse_hourly_series(raw, *point, timezone_name, variables)) samples = self._with_freezing_fallback_many( samples, chunk_points, start_date, end_date, timezone_name, source_type ) for position, sample in zip(chunk, samples): self._memory_cache[keys[position]] = (fetched_at, sample) results[position] = (sample, None) return results def resolve_many_models( self, points: list[tuple[float, float, float | None]], *, models: tuple[str, ...], start_date: date, end_date: date, timezone_name: str, variables: tuple[str, ...], source_type: WeatherSourceType = WeatherSourceType.FORECAST, freshness: dict | None = None, ) -> list[tuple[dict[str, TemporalWeatherSample | None] | None, str | None]]: """Named models side by side (step M1, docs/design_reference/ weather_start_models_spike_v0_1.md section 2): one multi-location request asks for every model at once (``models=a,b,c``); the provider answers each variable once per model, named ``_`` (unsuffixed when only one model is asked for). Per point: ({model: sample}, None), each sample carrying its model as the provenance of every series, or (None, failure_reason). A model that does not provide a variable here keeps the provider's empty series with the explicit reason MODEL_VARIABLE_NOT_PROVIDED -- never 0, never another model's value (no freezing-level fallback: that would put ICON's value under another model's name). Cached per point and model set (memory and disk), apart from resolve_many's single-model entries. The provider weighs each model as a separate call (spike section 2.2): N models cost N times the quota.""" models = _validated_models(models) joined = ",".join(models) results: list[tuple[dict[str, TemporalWeatherSample | None] | None, str | None]] = [ (None, None) ] * len(points) keys = [ _cache_key(lat, lon, ele, start_date, end_date, timezone_name) + (variables, ("models", models)) for lat, lon, ele in points ] def payload_path(point): return self._cache_path( self._payload_key( *point, start_date, end_date, timezone_name, source_type, joined, variables ) ) def split(raw, point): return { model: _model_sample(raw, model, models, point, timezone_name, variables) for model in models } if self._cells is not None: answers = self._cell_answers( points, model=joined, variables=variables, start_date=start_date, end_date=end_date, timezone_name=timezone_name, source_type=source_type, freshness=freshness, ) return [ (split(raw, point), None) if raw else (None, failure) for point, (raw, failure) in zip(points, answers) ] pending: list[int] = [] for position, (point, key) in enumerate(zip(points, keys)): cached = self._memory_cache.get(key) if cached is not None and self._is_fresh(cached[0]): results[position] = (cached[1], None) continue path = payload_path(point) if path.exists(): raw = json.loads(path.read_text(encoding="utf-8")) if self._is_fresh(raw.get(FETCHED_AT_KEY)): by_model = split(raw, point) self._memory_cache[key] = (raw[FETCHED_AT_KEY], by_model) results[position] = (by_model, None) continue pending.append(position) for first in range(0, len(pending), BATCH_MAX_LOCATIONS): chunk = pending[first : first + BATCH_MAX_LOCATIONS] chunk_points = [points[position] for position in chunk] try: raws = self._fetch_batch( chunk_points, start_date, end_date, timezone_name, source_type, joined, variables, ) except ProviderRateLimited as limited: remaining = chunk if limited.period == "MINUTE" else pending[first:] for position in remaining: results[position] = (None, limited.reason_code) if limited.period == "MINUTE": continue break except RuntimeError: for position in chunk: results[position] = (None, "PROVIDER_NETWORK_FAILURE") continue fetched_at = self._clock() for position, point, raw in zip(chunk, chunk_points, raws): if raw is None or "error" in raw: continue raw[FETCHED_AT_KEY] = fetched_at path = payload_path(point) _keep_on_disk(path, raw) by_model = split(raw, point) self._memory_cache[keys[position]] = (fetched_at, by_model) results[position] = (by_model, None) return results def _with_freezing_fallback_many( self, samples, points, start_date, end_date, timezone_name, source_type ): """The same regional freezing-level fallback as ``_resolve_uncached`` (ICON, labelled with the model), for every sample that asked for the freezing level and got none -- one batched request for all of them.""" needing = [ position for position, sample in enumerate(samples) if sample is not None and FREEZING_LEVEL_VARIABLE in sample.variables_by_time and not _has_any_value(sample.variables_by_time[FREEZING_LEVEL_VARIABLE]) ] if not needing: return samples samples = list(samples) try: raws = self._fetch_batch( [points[position] for position in needing], start_date, end_date, timezone_name, source_type, FREEZING_LEVEL_FALLBACK_MODEL, (FREEZING_LEVEL_VARIABLE,), ) except RuntimeError: for position in needing: samples[position] = _without_freezing_level( samples[position], FREEZING_LEVEL_FALLBACK_FETCH_FAILED ) return samples for position, raw in zip(needing, raws): sample = samples[position] series = _aligned_fallback_series(sample, raw, timezone_name) if not _has_any_value(series): samples[position] = _without_freezing_level( sample, FREEZING_LEVEL_NOT_PROVIDED_FOR_REGION ) continue unit = (raw or {}).get("hourly_units", {}).get(FREEZING_LEVEL_VARIABLE) samples[position] = replace( sample, base=replace( sample.base, units={**sample.base.units, FREEZING_LEVEL_VARIABLE: unit} ), variables_by_time={**sample.variables_by_time, FREEZING_LEVEL_VARIABLE: series}, variable_models={ **sample.variable_models, FREEZING_LEVEL_VARIABLE: FREEZING_LEVEL_FALLBACK_MODEL, }, ) return samples # --- AV-048: the shared grid-cell cache -------------------------------- def _cell_answers( self, points, *, model, variables, start_date, end_date, timezone_name, source_type, freshness=None, ) -> list[tuple[dict[str, Any] | None, str | None]]: """(raw answer cut to [start_date, end_date], failure) per point, from the cell cache: fresh -- as it is; stale (past the model's TTL, under a day old) -- as it is, refreshed in the background when there is a refresher; missing, too old or not covering the dates -- fetched now (each cell once, however many points share it).""" endpoint = OpenMeteoProvider._endpoint(source_type) cells = [grid_cell(lat, lon, ele, model) for lat, lon, ele in points] keys = [ cell_key( endpoint=endpoint, model=model, variables=variables, cell=cell, timezone_name=timezone_name, ) for cell in cells ] cell_of = dict(zip(keys, cells)) unique = list(dict.fromkeys(keys)) cached = self._cells.get_many(unique) # The model's TTL -- or longer inside the prefetch (AV-063). now = self._clock() served: dict[str, tuple[dict | None, str | None, float | None]] = {} to_fetch, to_refresh = [], [] for key in unique: entry = cached.get(key) covers = entry is not None and ( entry.start_date <= start_date and entry.end_date >= end_date ) age = now - entry.fetched_at if entry is not None else None # AV-051: the freshness of the model behind this very cell. fresh_for = effective_ttl(model, cell_of[key][0], cell_of[key][1]) if covers and age < fresh_for: served[key] = (entry.raw, None, entry.fetched_at) elif covers and age < MAX_STALE_SECONDS and self._refresher is not None: served[key] = (entry.raw, None, entry.fetched_at) to_refresh.append(key) else: to_fetch.append(key) served_from_cache = len(served) if served_from_cache: # AV-052: what the cache answered instead of a request. service_usage.record_cache_hit( "open_meteo", requests=served_from_cache, weight=call_weight( served_from_cache, len(model.split(",")) if model else 1, len(variables), (end_date - start_date).days + 1, ), ) if to_fetch: served.update( self._fetch_cells( to_fetch, cell_of, cached, model, variables, start_date, end_date, timezone_name, source_type, ) ) refreshing = bool(to_refresh) and self._schedule_refresh( to_refresh, cell_of, cached, model, variables, start_date, end_date, timezone_name, source_type, ) if freshness is not None: fetched = [at for _, failure, at in served.values() if at is not None] if fetched: oldest = min(fetched) freshness["oldest_fetched_at"] = min( freshness.get("oldest_fetched_at", oldest), oldest ) freshness["stale_cells"] = freshness.get("stale_cells", 0) + len(to_refresh) freshness["refreshing"] = freshness.get("refreshing", False) or refreshing freshness["cells"] = freshness.get("cells", 0) + len(unique) freshness["fetched_cells"] = freshness.get("fetched_cells", 0) + len(to_fetch) return [(_trim_raw(served[key][0], start_date, end_date), served[key][1]) for key in keys] def _fetch_cells( self, keys, cell_of, cached, model, variables, start_date, end_date, timezone_name, source_type, ) -> dict[str, tuple[dict[str, Any] | None, str | None, float | None]]: """Fetch these cells now, BATCH_MAX_LOCATIONS a request, and keep the answers. The dates asked for grow to what the cells' earlier answers covered (today's 72 hours and a trip's days served by one answer), at most 17 days; failures are never kept.""" today = datetime.now(ZoneInfo(timezone_name)).date() low, high = start_date, end_date for key in keys: entry = cached.get(key) if entry is not None and self._clock() - entry.fetched_at < MAX_STALE_SECONDS: low = min(low, max(entry.start_date, today - timedelta(days=1))) high = max(high, entry.end_date) # AV-051: grown to at most FETCH_MAX_DAYS -- Open-Meteo weighs more # than 14 days fractionally more (16 days = x1.14); only a range # asked for that is itself longer goes beyond. if (high - low).days + 1 > max(FETCH_MAX_DAYS, (end_date - start_date).days + 1): low, high = start_date, end_date results: dict[str, tuple[dict[str, Any] | None, str | None, float | None]] = {} models_count = len(model.split(",")) if model else 1 for first in range(0, len(keys), BATCH_MAX_LOCATIONS): batch = keys[first : first + BATCH_MAX_LOCATIONS] try: # AV-052: the request's quota weight for the usage counter. with service_usage.request_weight( call_weight(len(batch), models_count, len(variables), (high - low).days + 1) ): raws = self._fetch_batch( [cell_of[key] for key in batch], low, high, timezone_name, source_type, model, variables, bucketed=True, ) except ServiceBudgetExceeded as refused: # AV-052: our own threshold, not the provider's -- said as such. if self._log is not None: self._log(f"Open-Meteo not asked: {refused.level} budget ({refused.feature})") for key in keys[first:]: results[key] = (None, refused.reason_code, None) break except ProviderRateLimited as limited: self._note_failure( "RATE_LIMITED", call_weight(len(batch), models_count, len(variables), (high - low).days + 1), limited.period, limited.provider_reason, ) remaining = batch if limited.period == "MINUTE" else keys[first:] for key in remaining: results[key] = (None, limited.reason_code, None) if limited.period == "MINUTE": continue break except RuntimeError as error: self._note_failure( "ERROR", call_weight(len(batch), models_count, len(variables), (high - low).days + 1), None, str(error.__cause__ or error), ) for key in batch: results[key] = (None, "PROVIDER_NETWORK_FAILURE", None) continue fetched_at = self._clock() weight = call_weight(len(batch), models_count, len(variables), (high - low).days + 1) day = self._cells.record_calls(weight) self._cells.record_event(provider="open_meteo", outcome="OK", weight=weight) if self._log is not None: self._log( f"Open-Meteo: {len(batch)} locations ({model or 'best_match'}, " f"{low}..{high}) -- today {day['requests']} requests, " f"{day['weight']:.0f} of {day['limit']:.0f} calls" ) entries = [] for key, raw in zip(batch, raws): if raw is None or "error" in raw: results[key] = (None, None, None) continue entries.append((key, low, high, fetched_at, raw)) results[key] = (raw, None, fetched_at) self._cells.put_many(entries) return results def _near_limit(self) -> bool: usage = self._cells.calls_today() return usage["weight"] >= MET_FALLBACK_THRESHOLD * usage["limit"] def _met_norway_for_failures( self, results, points, start_date, end_date, timezone_name, variables ): """AV-051: the points Open-Meteo could not answer (a limit, a network failure, not asked), from MET Norway -- one request per grid cell, at most MET_FALLBACK_MAX_LOCATIONS; the variables MET Norway publishes, the rest absent with SOURCE_DOES_NOT_PROVIDE.""" from mountain_twin.weather.met_norway import ( MAPPING, SOURCE_DOES_NOT_PROVIDE, normalised, ) wanted = [ position for position, (sample, failure) in enumerate(results) if sample is None and failure in FALLBACK_FAILURES ] # AV-052: the background (prefetch, refresh) never moves to MET Norway. if ( not wanted or service_usage.current_feature("open_meteo") in service_usage.BACKGROUND_FEATURES ): return results by_cell: dict[tuple, TemporalWeatherSample | None] = {} results = list(results) for position in wanted: latitude, longitude, elevation = points[position] cell = grid_cell(latitude, longitude, elevation, None) if cell not in by_cell: if len(by_cell) >= MET_FALLBACK_MAX_LOCATIONS: continue try: answer = self._fallback.forecast(*cell) except RuntimeError as error: by_cell[cell] = None if self._cells is not None: self._cells.record_event( provider="met_norway", outcome="ERROR", weight=1.0, message=str(error) ) if self._log is not None: self._log(f"MET Norway failed: {error}") continue if self._cells is not None: self._cells.record_event(provider="met_norway", outcome="OK", weight=1.0) raw = normalised( answer, timezone_name=timezone_name, start_date=start_date, end_date=end_date, variables=variables, ) sample = _parse_hourly_series(raw, *cell, timezone_name, variables) if sample is not None: reasons = dict(sample.base.missing_variable_reasons) reasons.update( {v: SOURCE_DOES_NOT_PROVIDE for v in variables if v not in MAPPING} ) sample = replace( sample, base=replace(sample.base, missing_variable_reasons=reasons), variable_models={v: "met_norway" for v in variables if v in MAPPING}, ) by_cell[cell] = sample sample = by_cell.get(cell) if sample is not None: results[position] = (sample, None) return results def _note_failure(self, outcome, weight, period, message) -> None: """AV-051: a refused or failed request -- kept with its limit and the provider's words, logged, remembered for the page's message.""" if period is not None: now = self._clock() self.last_rate_limit = { "period": period, "reason_code": RATE_LIMIT_REASON_BY_PERIOD[period], "provider_reason": message, "at": now, "until": rate_limit_until(period, now), } if self._cells is not None: self._cells.record_event( provider="open_meteo", outcome=outcome, weight=weight, limit_period=period, message=message, ) if self._log is not None: self._log( f"Open-Meteo {outcome}{f' ({period})' if period else ''}: {message or 'no reason given'}" ) def rate_limit_document(self) -> dict | None: """The newest limit while it lasts (None after it is lifted).""" limit = getattr(self, "last_rate_limit", None) if not limit or self._clock() >= limit["until"]: return None return { "period": limit["period"], "reason_code": limit["reason_code"], "provider_reason": limit["provider_reason"], "until": datetime.fromtimestamp(limit["until"], timezone.utc).isoformat(), } def _schedule_refresh( self, keys, cell_of, cached, model, variables, start_date, end_date, timezone_name, source_type, ) -> bool: """Stale-while-revalidate: fetch these cells again in the background (each cell once at a time, however many requests saw it stale).""" with self._in_flight_lock: keys = [key for key in keys if key not in self._in_flight] self._in_flight.update(keys) if not keys: return True # AV-052: the refresh belongs to whoever asked -- the prefetch's own # stale cells are the prefetch's cost, not a page's. asked_by = "PREFETCH" if current_call_source() == "PREFETCH" else "REFRESH" def refresh(): try: with call_source(asked_by): self._fetch_cells( keys, cell_of, cached, model, variables, start_date, end_date, timezone_name, source_type, ) except Exception as error: # AV-051: a failed refresh leaves a trace if self._log is not None: self._log( f"Open-Meteo background refresh failed: {type(error).__name__}: {error}" ) finally: with self._in_flight_lock: self._in_flight.difference_update(keys) self._refresher(refresh) return True def _with_freezing_fallback_cells( self, samples, points, start_date, end_date, timezone_name, source_type, freshness ): """_with_freezing_fallback_many through the cell cache.""" needing = [ position for position, sample in enumerate(samples) if sample is not None and FREEZING_LEVEL_VARIABLE in sample.variables_by_time and not _has_any_value(sample.variables_by_time[FREEZING_LEVEL_VARIABLE]) ] if not needing: return samples samples = list(samples) answers = self._cell_answers( [points[position] for position in needing], model=FREEZING_LEVEL_FALLBACK_MODEL, variables=(FREEZING_LEVEL_VARIABLE,), start_date=start_date, end_date=end_date, timezone_name=timezone_name, source_type=source_type, freshness=freshness, ) for position, (raw, failure) in zip(needing, answers): sample = samples[position] if raw is None and failure: samples[position] = _without_freezing_level( sample, FREEZING_LEVEL_FALLBACK_FETCH_FAILED ) continue series = _aligned_fallback_series(sample, raw, timezone_name) if not _has_any_value(series): samples[position] = _without_freezing_level( sample, FREEZING_LEVEL_NOT_PROVIDED_FOR_REGION ) continue unit = (raw or {}).get("hourly_units", {}).get(FREEZING_LEVEL_VARIABLE) samples[position] = replace( sample, base=replace( sample.base, units={**sample.base.units, FREEZING_LEVEL_VARIABLE: unit} ), variables_by_time={**sample.variables_by_time, FREEZING_LEVEL_VARIABLE: series}, variable_models={ **sample.variable_models, FREEZING_LEVEL_VARIABLE: FREEZING_LEVEL_FALLBACK_MODEL, }, ) return samples def _fetch_batch( self, points, start_date, end_date, timezone_name, source_type, model, variables, bucketed: bool = False, ) -> list[dict[str, Any] | None]: """One multi-location request; the provider answers a list (one object per location, in order) -- or a single object for one location. A response that does not line up with the request is a failure, never shifted onto the wrong points.""" params = { "latitude": ",".join(str(lat) for lat, _, _ in points), "longitude": ",".join(str(lon) for _, lon, _ in points), # "nan": the provider's own terrain height (never sent by the # Weather tab, which does not query points without elevation). # bucketed: the cell's own elevation bucket, sent as it is. "elevation": ",".join( "nan" if ele is None else str(ele if bucketed else _bucketed_elevation(ele)) for _, _, ele in points ), "hourly": ",".join(variables), "timezone": timezone_name, "start_date": start_date.isoformat(), "end_date": end_date.isoformat(), } if model is not None: params["models"] = model raw = self._fetch_json(f"{OpenMeteoProvider._endpoint(source_type)}?{urlencode(params)}") raws = raw if isinstance(raw, list) else [raw] if len(raws) != len(points): raise RuntimeError("WEATHER_PROVIDER_BATCH_MISMATCH") return raws @staticmethod def _payload_key( latitude, longitude, elevation_m, start_date, end_date, timezone_name, source_type, model, variables, ) -> dict[str, Any]: payload_key = { "endpoint": OpenMeteoProvider._endpoint(source_type), "latitude": latitude, "longitude": longitude, "start_date": start_date.isoformat(), "end_date": end_date.isoformat(), "timezone": timezone_name, "hourly_variables": list(variables), "elevation": _bucketed_elevation(elevation_m), } if model is not None: payload_key["models"] = model return payload_key def _cache_path(self, payload_key: dict[str, Any]) -> Path: disk_key = hashlib.sha256( json.dumps(payload_key, sort_keys=True, separators=(",", ":")).encode() ).hexdigest() directory = self._provider.cache_directory if self._opener is urlopen and offline_mode() is not None: # AV-045: offline mode neither reads nor fills the real cache (a # development server's): this process's own scratch directory. directory = offline_cache_directory() return directory / f"live_point_{disk_key}.json" @staticmethod def _build_url( latitude, longitude, elevation_m, start_date, end_date, timezone_name, source_type, model: str | None = None, variables: tuple[str, ...] = LIVE_HOURLY_VARIABLES, ) -> str: params = { "latitude": latitude, "longitude": longitude, "hourly": ",".join(variables), "timezone": timezone_name, "start_date": start_date.isoformat(), "end_date": end_date.isoformat(), } # Without it Open-Meteo downscales to its own terrain height for the # grid cell, which can be far from the route point: at 45.8326 N # 6.8652 E (Chamonix valley side) it used 4792 m -- the Mont Blanc # cell -- and returned -2.7 C where 2500 m gave 12.3 C (2026-09-27). bucketed_elevation = _bucketed_elevation(elevation_m) if bucketed_elevation is not None: params["elevation"] = bucketed_elevation if model is not None: params["models"] = model return f"{OpenMeteoProvider._endpoint(source_type)}?{urlencode(params)}" # A model name as Open-Meteo spells it (e.g. "icon_seamless", "ecmwf_ifs025"). _MODEL_NAME = re.compile(r"^[a-z0-9_]+$") MODEL_VARIABLE_NOT_PROVIDED = "MODEL_VARIABLE_NOT_PROVIDED" def _validated_models(models) -> tuple[str, ...]: models = tuple(models) if not models: raise ValueError("at least one weather model is needed") if len(set(models)) != len(models): raise ValueError("a weather model is named twice") for model in models: if not isinstance(model, str) or not _MODEL_NAME.match(model): raise ValueError(f"not a weather model name: {model!r}") return models def _model_sample( raw: dict[str, Any], model: str, models: tuple[str, ...], point: tuple[float, float, float | None], timezone_name: str, variables: tuple[str, ...], ) -> TemporalWeatherSample | None: """One model's series out of a multi-model response, under the plain variable names and labelled with the model.""" suffix = "" if len(models) == 1 else f"_{model}" hourly, units = raw.get("hourly", {}), raw.get("hourly_units", {}) single = { **{key: value for key, value in raw.items() if key not in ("hourly", "hourly_units")}, "hourly": { "time": hourly.get("time", []), **{ variable: hourly[f"{variable}{suffix}"] for variable in variables if f"{variable}{suffix}" in hourly }, }, "hourly_units": { variable: units.get(f"{variable}{suffix}") for variable in variables if f"{variable}{suffix}" in units }, } sample = _parse_hourly_series(single, *point, timezone_name, variables) if sample is None: return None not_provided = { variable: MODEL_VARIABLE_NOT_PROVIDED for variable, series in sample.variables_by_time.items() if not _has_any_value(series) } return replace( sample, base=replace( sample.base, missing_variable_reasons={**sample.base.missing_variable_reasons, **not_provided}, ), variable_models={variable: model for variable in variables}, ) def _trim_raw(raw: dict[str, Any] | None, start_date: date, end_date: date): """A cached answer cut to the dates asked for (it may cover more): the hourly arrays kept for those local dates only, nothing else touched.""" if raw is None: return None hourly = raw.get("hourly") or {} times = hourly.get("time") or [] first, last = start_date.isoformat(), end_date.isoformat() keep = [i for i, value in enumerate(times) if first <= value[:10] <= last] if len(keep) == len(times): return raw return { **raw, "hourly": { name: [series[i] for i in keep] if isinstance(series, list) else series for name, series in hourly.items() }, } def _has_any_value(series) -> bool: return bool(series) and any(value is not None for value in series) def _without_freezing_level(sample: TemporalWeatherSample, reason: str) -> TemporalWeatherSample: """The default series stays as the provider sent it (all nulls); the reason why there is no value is made explicit next to it.""" base = replace( sample.base, missing_variable_reasons={ **sample.base.missing_variable_reasons, FREEZING_LEVEL_VARIABLE: reason, }, ) return replace(sample, base=base) def _aligned_fallback_series( sample: TemporalWeatherSample, fallback_raw: dict[str, Any] | None, timezone_name: str ) -> tuple[float | int | None, ...] | None: """The fallback model's freezing level on exactly the sample's hours, or None if the response does not line up (never shifted or resampled).""" if fallback_raw is None: return None hourly = fallback_raw.get("hourly", {}) zone = ZoneInfo(timezone_name) times = tuple( datetime.fromisoformat(value).replace(tzinfo=zone) for value in hourly.get("time", []) ) series = hourly.get(FREEZING_LEVEL_VARIABLE) if times != sample.times or series is None or len(series) != len(times): return None unit = fallback_raw.get("hourly_units", {}).get(FREEZING_LEVEL_VARIABLE) return tuple( None if value is None or _invalid_value_reason(FREEZING_LEVEL_VARIABLE, value, unit) else value for value in series ) # AV-048: the points of one request share their hour axis; it is parsed once # for all of them (the last few axes kept -- tuples, immutable). _TIMES_CACHE: dict[tuple, tuple[datetime, ...]] = {} _TIMES_CACHE_LOCK = threading.Lock() def _parsed_times(times_raw: tuple[str, ...], timezone_name: str) -> tuple[datetime, ...]: key = (times_raw, timezone_name) with _TIMES_CACHE_LOCK: cached = _TIMES_CACHE.get(key) if cached is not None: return cached zone = ZoneInfo(timezone_name) times = tuple(datetime.fromisoformat(value).replace(tzinfo=zone) for value in times_raw) with _TIMES_CACHE_LOCK: if len(_TIMES_CACHE) >= 32: _TIMES_CACHE.clear() _TIMES_CACHE[key] = times return times def _parse_hourly_series( raw: dict[str, Any], latitude: float, longitude: float, elevation_m: float | None, timezone_name: str, variables: tuple[str, ...] = LIVE_HOURLY_VARIABLES, ) -> TemporalWeatherSample | None: hourly = raw.get("hourly", {}) times_raw = hourly.get("time", []) if not times_raw: return None times = _parsed_times(tuple(times_raw), timezone_name) units = raw.get("hourly_units", {}) variables_by_time: dict[str, tuple[float | int | None, ...]] = {} missing_variable_reasons: dict[str, str] = {} variable_semantics: dict[str, dict[str, Any]] = {} for variable in variables: semantics = weather_variable_semantics(variable) if semantics: variable_semantics[variable] = semantics series = hourly.get(variable) if series is None: missing_variable_reasons[variable] = "PROVIDER_VARIABLE_UNSUPPORTED_OR_ABSENT" continue resolved: list[float | int | None] = [] for value in series: if value is None: resolved.append(None) continue invalid_reason = _invalid_value_reason(variable, value, units.get(variable)) resolved.append(None if invalid_reason else value) variables_by_time[variable] = tuple(resolved) provider_elevation = raw.get("elevation") base = RouteWeatherSample( sample_id=f"live_point_{latitude:.4f}_{longitude:.4f}", point_index=0, route_distance_m=0.0, latitude=latitude, longitude=longitude, route_elevation_m=elevation_m, provider_elevation_m=provider_elevation, elevation_difference_m=( elevation_m - provider_elevation if elevation_m is not None and provider_elevation is not None else None ), selection_reasons=("live_point_query",), variables={}, units={variable: units.get(variable) for variable in variables}, missing_variable_reasons=missing_variable_reasons, variable_semantics=variable_semantics, ) return TemporalWeatherSample( base=base, timezone=timezone_name, times=times, variables_by_time=variables_by_time )