"""Background weather prefetch (AV-048, docs/reports/AV-048_raport.md section 4): every PREFETCH_INTERVAL_HOURS the local server builds the Weather tab's default document for the *active* Journeys, so opening the tab finds the shared cell cache (mountain_twin/weather/forecast_cache.py) warm. Active: a saved Journey with a route, not marked completed, that was opened in the last 7 days, or whose plan is under way or starts within the forecast's reach. The default model only -- the model comparison stays on request (it weighs three times as much). The provider's free quota is shared with the person using the app: before each Journey the day's call count (the cache's own counter) is checked, and the prefetch pauses until the next cycle once BUDGET_FRACTION of the daily limit is used. Each cycle and each pause is logged. AV-063: a restart costs nothing new. The prefetch leaves answers younger than PREFETCH_FRESH_SECONDS alone (forecast_cache.fresh_for_at_least), and the first cycle after a start waits for the next regular one -- the last cycle's end is kept in the cache file -- at least INITIAL_DELAY_SECONDS so the server's start is never slowed. Tests never start it (scripts/serve_visual_prototype.py does, only from its command line, and WEATHER_PREFETCH=0 turns it off). """ from __future__ import annotations import os import threading import time from dataclasses import dataclass from datetime import datetime, timedelta from typing import Callable, Iterable from mountain_twin.weather.forecast_cache import ( PREFETCH_FRESH_SECONDS, ForecastCellCache, call_source, fresh_for_at_least, ) PREFETCH_ENVIRONMENT = "WEATHER_PREFETCH" INTERVAL_ENVIRONMENT = "WEATHER_PREFETCH_HOURS" DEFAULT_INTERVAL_HOURS = 3.0 INITIAL_DELAY_SECONDS = 120.0 RECENTLY_OPENED_SECONDS = 7 * 24 * 3600.0 # AV-051: a trip starting more than this far ahead is forecast by the global # models alone (the regional ones -- ICON-D2, AROME, ICON-EU -- reach 2-5 # days), which publish a new run every 6 h: refreshing it every 3 h fetched # the same run twice. REGIONAL_REACH = timedelta(days=5) GLOBAL_RUN_SECONDS = 6 * 3600.0 NEAR_REACH = timedelta(days=2) FAR_RUN_SECONDS = 12 * 3600.0 FORECAST_REACH = timedelta(days=16) BUDGET_FRACTION = 0.5 @dataclass(frozen=True) class PrefetchCandidate: journey_id: str completed: bool has_route: bool trip_start: datetime | None = None trip_end: datetime | None = None def prefetch_enabled(environ=os.environ) -> bool: """On unless WEATHER_PREFETCH=0 -- and never in offline mode (the tests: a background thread asking for weather has no place there).""" if (environ.get("ECHTRO_OFFLINE") or "").strip().lower() not in ("", "0", "false", "no", "off"): return False return (environ.get(PREFETCH_ENVIRONMENT) or "1").strip().lower() not in ( "0", "false", "no", "off", ) def prefetch_interval_seconds(environ=os.environ) -> float: """1-3 h is the sensible range (the default model's cycle is 1 h, the quota is daily); the value is clamped to [1, 24] h.""" try: hours = float(environ.get(INTERVAL_ENVIRONMENT) or DEFAULT_INTERVAL_HOURS) except ValueError: hours = DEFAULT_INTERVAL_HOURS return min(24.0, max(1.0, hours)) * 3600.0 def active_journeys( candidates: Iterable[PrefetchCandidate], recently_opened: Iterable[str], now: datetime ) -> list[str]: """AV-051 section 5 (AV-052): opened in the last 7 days AND within the forecast's reach -- under way, starting within it, or without a timed start (its tab shows the next hours); the most recently opened first. Never a completed one or one without a route. (AV-048 took either; the usage simulator showed trips nobody had opened refetched all day.)""" eligible = {c.journey_id: c for c in candidates if c.has_route and not c.completed} chosen = [] for journey_id in recently_opened: candidate = eligible.get(journey_id) if candidate is None: continue if candidate.trip_start is None: chosen.append(journey_id) continue end = candidate.trip_end or candidate.trip_start if end >= now and candidate.trip_start <= now + FORECAST_REACH: chosen.append(journey_id) return chosen def fresh_for(trip_start: datetime | None, now: datetime) -> float: """How old an answer the prefetch leaves alone for this trip (AV-052, from the usage simulator): under way, without a timed start or starting within NEAR_REACH -- 3 h (every regional run); within REGIONAL_REACH -- 6 h (the global models' run); further -- 12 h (twice a day: so far ahead a run changes less than the forecast's own uncertainty). Opening the trip refreshes it at its model's own pace anyway.""" if trip_start is None or trip_start <= now + NEAR_REACH: return PREFETCH_FRESH_SECONDS if trip_start <= now + REGIONAL_REACH: return GLOBAL_RUN_SECONDS return FAR_RUN_SECONDS class WeatherPrefetcher: """``candidates()`` lists the saved Journeys; ``warm(journey_id)`` builds one Journey's default Weather document (filling the cell cache). Both come from the server; this class only decides when and which.""" def __init__( self, *, cache: ForecastCellCache, candidates: Callable[[], Iterable[PrefetchCandidate]], warm: Callable[[str], object], interval_seconds: float, initial_delay_seconds: float = INITIAL_DELAY_SECONDS, log: Callable[[str], None] = lambda line: None, now: Callable[[], datetime] | None = None, ) -> None: self.cache = cache self.candidates = candidates self.warm = warm self.interval_seconds = interval_seconds self.initial_delay_seconds = initial_delay_seconds self.log = log self.now = now or (lambda: datetime.now().astimezone()) self._stop = threading.Event() self._thread: threading.Thread | None = None def over_budget(self) -> dict | None: usage = self.cache.calls_today() return usage if usage["weight"] >= BUDGET_FRACTION * usage["limit"] else None def run_once(self) -> dict: """One cycle: the active Journeys, warmed one by one until the budget says stop. Returns what was done (for the log and the tests).""" started = time.monotonic() candidates = list(self.candidates()) now = self.now() chosen = active_journeys(candidates, self.cache.opened_since(RECENTLY_OPENED_SECONDS), now) starts = {candidate.journey_id: candidate.trip_start for candidate in candidates} warmed, failed, paused = [], [], None for journey_id in chosen: paused = self.over_budget() if paused: self.log( f"Weather prefetch paused: {paused['weight']:.0f} of " f"{paused['limit']:.0f} Open-Meteo calls used today " f"(the limit is {BUDGET_FRACTION:.0%} for the prefetch)" ) break try: with ( fresh_for_at_least(fresh_for(starts.get(journey_id), now)), call_source("PREFETCH"), ): self.warm(journey_id) warmed.append(journey_id) except Exception as error: # one Journey's failure never stops the rest failed.append((journey_id, f"{type(error).__name__}: {error}")) if hasattr(self.cache, "note_prefetch"): self.cache.note_prefetch() usage = self.cache.calls_today() self.log( f"Weather prefetch: {len(warmed)} of {len(chosen)} active Journeys warmed " f"in {time.monotonic() - started:.1f} s; Open-Meteo today {usage['requests']} " f"requests, {usage['weight']:.0f} of {usage['limit']:.0f} calls" + (f"; failed: {failed}" if failed else "") ) return {"chosen": chosen, "warmed": warmed, "failed": failed, "paused": paused} def first_delay_seconds(self) -> float: """Until the cycle the last one scheduled (kept across restarts), never sooner than the initial delay.""" last = self.cache.last_prefetch() if hasattr(self.cache, "last_prefetch") else None due = 0.0 if last is None else last + self.interval_seconds - self.cache.clock() return max(self.initial_delay_seconds, due) def start(self) -> threading.Thread: def loop(): delay = self.first_delay_seconds() if delay > self.initial_delay_seconds: self.log(f"Weather prefetch: last cycle recent, next in {delay / 60:.0f} min") if self._stop.wait(delay): return while not self._stop.is_set(): try: self.run_once() except Exception as error: # the next cycle tries again self.log(f"Weather prefetch cycle failed: {type(error).__name__}: {error}") self.cache.prune() if self._stop.wait(self.interval_seconds): return self._thread = threading.Thread(target=loop, name="weather-prefetch", daemon=True) self._thread.start() return self._thread def stop(self) -> None: self._stop.set()