"""Satellite cloud imagery for the Storms / Precipitation tab (docs/design_reference/profile_storms_spike_v0_1.md section f). RainViewer discontinued its satellite frames (January 2026); the imagery here is EUMETSAT's public EUMETView WMS -- no key, no account, for personal and non-commercial use with the credit "©EUMETSAT [year]" (a debt to revisit before any commercial release, like RainViewer's terms). Two infrared layers: Meteosat at 0 degrees (Europe, Africa, the Atlantic; a new image every 10 minutes) where the place is well inside its disc, the global ring of geostationary satellites (every 3 hours) elsewhere. The server asks the layer's own capabilities for the time of its newest image (cached ~5 min for every viewer, retried, degraded to an explicit UNAVAILABLE) and derives a short run of PAST frames at that same interval (live-verified: EUMETSAT keeps hours of history on this WMS, not only the newest image) -- the same "past frames, newest last" shape mountain_twin.weather.radar already uses, so the browser drives both through the one player mechanism. The global ring's images are 3 hours apart -- a jump too big to read as an animation -- so it gets a single frame, an explicit product choice (docs/design_reference/ profile_storms_spike_v0_1.md section f.a), not a data gap: the browser loads the tiles directly from EUMETSAT either way. An observed image, not a forecast: thermal infrared, brighter = colder = higher cloud tops. Nothing here says what the weather will do. """ from __future__ import annotations import re import threading import time from datetime import datetime, timedelta, timezone from typing import Any, Callable from mountain_twin.offline import urlopen # AV-045: offline mode for tests EUMETVIEW_WMS_URL = "https://view.eumetsat.int/geoserver/ows" CLOUDS_CACHE_SECONDS = 300 CLOUDS_FETCH_ATTEMPTS = 2 CLOUDS_CAPTION = ( "Chmury: zdjęcie satelitarne w podczerwieni — im bielej, tym zimniej, " "czyli wyższe wierzchołki chmur; ciepła powierzchnia jest przezroczysta " "(niskie chmury i mgły bywają przez to słabo widoczne). Obserwacja, nie prognoza." ) # Meteosat at 0 degrees sees a disc centred on 0 N 0 E; towards its rim the # image is badly foreshortened. Inside these bounds its 10-minute imagery is # used, outside them the 3-hourly global ring. METEOSAT_0_MAX_ABS_LONGITUDE = 55.0 METEOSAT_0_MAX_ABS_LATITUDE = 70.0 LAYERS = { "METEOSAT_0_DEGREE": { "layer": "mtg_fd:ir105_hrfi", "title": "Meteosat (MTG) 0°, podczerwień 10,5 µm", "interval_minutes": 10, # Six frames = the last 50 minutes, a real animation. "frame_count": 6, }, "GEOSTATIONARY_RING": { "layer": "mumi:worldcloudmap_ir108", "title": "pierścień satelitów geostacjonarnych, podczerwień 10,8 µm", "interval_minutes": 180, # A single image: see the module docstring -- not a data gap. "frame_count": 1, }, } _TIME_DEFAULT = re.compile(r']*\sdefault="([^"]+)"') def coverage_for(latitude: float, longitude: float) -> str: inside = ( abs(longitude) <= METEOSAT_0_MAX_ABS_LONGITUDE and abs(latitude) <= METEOSAT_0_MAX_ABS_LATITUDE ) return "METEOSAT_0_DEGREE" if inside else "GEOSTATIONARY_RING" class EumetsatCloudProvider: """The most recent infrared frames over a place, as WMS layer descriptions the map can draw and animate (one frame where the source itself is only worth showing once -- never a pretended animation).""" def __init__( self, fetch_text: Callable[[str], str] | None = None, clock: Callable[[], float] = time.time, ): self._fetch_text = fetch_text or _fetch_text self._clock = clock self._lock = threading.Lock() self._cached: dict[str, tuple[float, dict[str, Any]]] = {} def frame_document(self, latitude: float, longitude: float) -> dict[str, Any]: """The document shape mirrors mountain_twin.weather.radar's: a ``frames`` list (oldest first, newest last) the browser plays through the same player as the precipitation radar.""" coverage = coverage_for(latitude, longitude) with self._lock: now = self._clock() cached = self._cached.get(coverage) if cached and now - cached[0] < CLOUDS_CACHE_SECONDS: return cached[1] document = self._fresh_document(coverage) # A failure is cached briefly too: every viewer does not retry it. self._cached[coverage] = (now, document) return document def _fresh_document(self, coverage: str) -> dict[str, Any]: spec = LAYERS[coverage] workspace, name = spec["layer"].split(":") url = ( f"https://view.eumetsat.int/geoserver/{workspace}/{name}/ows" "?service=WMS&version=1.3.0&request=GetCapabilities" ) last_error = None for _ in range(CLOUDS_FETCH_ATTEMPTS): try: match = _TIME_DEFAULT.search(self._fetch_text(url)) if not match: raise ValueError("the layer's capabilities name no image time") newest = datetime.fromisoformat(match.group(1).replace("Z", "+00:00")) step = timedelta(minutes=spec["interval_minutes"]) frame_times = [ newest - step * age for age in range(spec["frame_count"] - 1, -1, -1) ] return { "state": "AVAILABLE", "provider": "EUMETSAT", "kind": "OBSERVED", "image": "THERMAL_INFRARED", "coverage": coverage, "layer": spec["layer"], "layer_title": spec["title"], "interval_minutes": spec["interval_minutes"], "wms_url": EUMETVIEW_WMS_URL, "caption": CLOUDS_CAPTION, "attribution": { "text": f"©EUMETSAT {newest.year}", "url": "https://view.eumetsat.int/", }, # The page draws the image white on clear (AV-024), so # the layer can be nearly opaque without hiding the map. "opacity": 0.9, "frames": [ { "time": frame_time.astimezone(timezone.utc).isoformat(), # The same instant as the WMS TIME parameter spells it. "wms_time": frame_time.astimezone(timezone.utc).strftime( "%Y-%m-%dT%H:%M:%SZ" ), } for frame_time in frame_times ], } except Exception as error: # network, HTTP or shape last_error = error return { "state": "UNAVAILABLE", "reason": "PROVIDER_UNAVAILABLE", "detail": type(last_error).__name__ if last_error else None, "provider": "EUMETSAT", "coverage": coverage, } def disabled_clouds_document() -> dict[str, Any]: return {"state": "DISABLED", "reason": "RADAR_DISABLED_BY_CONFIGURATION"} def _fetch_text(url: str) -> str: with urlopen(url, timeout=10) as response: # noqa: S310 - fixed https URL return response.read().decode("utf-8", errors="replace") __all__ = [ "CLOUDS_CAPTION", "EUMETVIEW_WMS_URL", "EumetsatCloudProvider", "coverage_for", "disabled_clouds_document", ]