Fallback to GFS wave dataset and improve UI messaging

README.md

@@ -8,12 +8,12 @@ pinned: false
 license: cc-by-4.0
 ---
 
-## GEFS Wave Control Member Downloader
+## Wave Dataset Downloader
 
-This Space now provides a Gradio dashboard that keeps a local cache of the latest GEFS Wave control member (`c00`, “wave 0”) global 0.25° GRIB2 dataset from NOAA NOMADS. Use the interface to:
+This Space now provides a Gradio dashboard that keeps a local cache of the latest NOAA wave datasets. It first tries the GEFS Wave control member (`c00`, “wave 0”) via the Azure mirror and NOMADS; if no ensemble cycle is available, it automatically falls back to the operational GFS Wave global 0.25° product served from NOMADS and the NCEP Polar site. Use the interface to:
 
 - Inspect the current downloader status and metadata for the newest available cycle.
 - Trigger a refresh to fetch a more recent cycle if one has been published.
 - Download a `.tar.gz` archive containing every `global.0p25` GRIB2 file for that cycle.
 
-Data is stored under `data/gefswave` inside the Space. Override this path with the `GEFS_WAVE_DATA_DIR` environment variable if needed. The downloader searches up to five days back to locate the newest complete cycle.
+Data is stored under `data/gefswave` (configurable via `GEFS_WAVE_DATA_DIR`). You can optionally supply an Azure SAS token with `GEFS_AZURE_SAS_TOKEN`; otherwise, the downloader will request one from the Planetary Computer API when needed. Logs in the UI show which source was used and how many cycles were checked.

app.py

@@ -55,6 +55,8 @@ def format_status(status: Dict, notice: Optional[str] = None) -> Tuple[Dict, str
     if "message" in status:
         lines.append(f"- Message: {status['message']}")
     if latest:
+        if dataset := latest.get("product_label"):
+            lines.append(f"- Dataset: `{dataset}`")
         lines.extend(
             [
                 f"- Latest cycle: `{latest.get('cycle', 'N/A')}`",

gefs_wave.py

@@ -11,13 +11,11 @@ from typing import Dict, Iterable, List, Optional
 import requests
 
 
-NOMADS_URL = "https://nomads.ncep.noaa.gov/pub/data/nccf/com/gens/prod"
-AZURE_URL = "https://noaagefs.blob.core.windows.net/gefs"
-BASE_URLS = [AZURE_URL, NOMADS_URL]
 AZURE_SAS_ENDPOINT = "https://planetarycomputer.microsoft.com/api/sas/v1/token/noaagefs/gefs"
-FORECAST_HOURS: List[int] = list(range(0, 387, 3))
+FORECAST_HOURS_GEFS: List[int] = list(range(0, 387, 3))
+FORECAST_HOURS_GFS: List[int] = list(range(0, 387, 3))
 STATE_FILENAME = "state.json"
-TARBALL_FILENAME = "gefswave-wave0-latest.tar.gz"
+
 
 logger = logging.getLogger("gefs_wave")
 if not logger.handlers:
@@ -28,6 +26,53 @@ logger.setLevel(logging.INFO)
 logger.propagate = False
 
 
+@dataclass
+class Endpoint:
+    base_url: str
+    mode: str = "dated"  # "dated" or "latest"
+    needs_sas: bool = False
+    label: Optional[str] = None
+
+
+@dataclass
+class WaveProduct:
+    key: str
+    label: str
+    directory_template: str
+    filename_template: str
+    endpoints: List[Endpoint]
+    forecast_hours: List[int]
+    max_days_back: int
+
+
+PRODUCTS: List[WaveProduct] = [
+    WaveProduct(
+        key="gefswave_c00",
+        label="GEFS Wave Control (c00)",
+        directory_template="gefs.{date}/{cycle}/wave/gridded",
+        filename_template="gefswave.t{cycle}z.c00.global.0p25.f{hour:03d}.grib2",
+        endpoints=[
+            Endpoint("https://noaagefs.blob.core.windows.net/gefs", needs_sas=True, label="Azure GEFS mirror"),
+            Endpoint("https://nomads.ncep.noaa.gov/pub/data/nccf/com/gens/prod", label="NOMADS GEFS"),
+        ],
+        forecast_hours=FORECAST_HOURS_GEFS,
+        max_days_back=9,
+    ),
+    WaveProduct(
+        key="gfswave_global",
+        label="GFS Wave Global 0.25°",
+        directory_template="gfs.{date}/{cycle}/wave/gridded",
+        filename_template="gfswave.t{cycle}z.global.0p25.f{hour:03d}.grib2",
+        endpoints=[
+            Endpoint("https://nomads.ncep.noaa.gov/pub/data/nccf/com/wave/prod", label="NOMADS GFS"),
+            Endpoint("https://polar.ncep.noaa.gov/waves/latest_run", mode="latest", label="NOAA Polar latest_run"),
+        ],
+        forecast_hours=FORECAST_HOURS_GFS,
+        max_days_back=5,
+    ),
+]
+
+
 @dataclass
 class Cycle:
     date: str  # YYYYMMDD
@@ -37,17 +82,17 @@ class Cycle:
     def label(self) -> str:
         return f"{self.date} {self.cycle}Z"
 
-    @property
-    def directory(self) -> str:
-        return f"gefs.{self.date}/{self.cycle}"
-
 
 class WaveDownloadError(Exception):
     """Raised when a download or discovery step fails."""
 
 
+class RetryableAuthError(WaveDownloadError):
+    """Raised when an auth token should be refreshed and retried."""
+
+
 class WaveDownloader:
-    """Handles discovery and download of the GEFS wave control-member dataset."""
+    """Handles discovery and download of wave datasets."""
 
     def __init__(self, data_root: Path, session: Optional[requests.Session] = None) -> None:
         self.data_root = data_root
@@ -57,23 +102,39 @@ class WaveDownloader:
         self._azure_token_checked = False
 
     def ensure_latest_cycle(self) -> Dict:
-        """Ensure the latest available control-member dataset is present locally."""
-        logger.info("Ensuring latest GEFS wave control-member cycle for %s", self.data_root)
-        cycle = self._find_latest_cycle()
+        """Ensure the latest available dataset (GEFS first, then GFS) is cached locally."""
+        errors: List[str] = []
+        for product in PRODUCTS:
+            try:
+                return self._ensure_product(product)
+            except WaveDownloadError as exc:
+                logger.warning("Failed to refresh %s: %s", product.label, exc)
+                errors.append(f"{product.label}: {exc}")
+
+        raise WaveDownloadError("All configured products failed. " + " | ".join(errors))
+
+    def _ensure_product(self, product: WaveProduct) -> Dict:
+        logger.info("Ensuring latest %s dataset in %s", product.label, self.data_root)
+        cycle = self._find_latest_cycle(product)
         if not cycle:
-            logger.error("No GEFS wave cycle found in the recent window.")
-            raise WaveDownloadError("Could not locate a recent GEFS wave cycle.")
+            raise WaveDownloadError("Could not locate a recent cycle.")
 
         state = self._read_state()
-        if state and state.get("cycle") == cycle.label and self._cycle_complete(cycle):
-            logger.info("Cycle %s already cached; using existing files.", cycle.label)
+        if (
+            state
+            and state.get("product") == product.key
+            and state.get("cycle") == cycle.label
+            and self._cycle_complete(product, cycle)
+        ):
+            logger.info("Cycle %s already cached for %s.", cycle.label, product.label)
             return state
 
-        logger.info("Downloading cycle %s", cycle.label)
-        files = self._download_cycle(cycle)
-        tarball_path = self._build_tarball(cycle, files)
-        logger.info("Packaged cycle %s into tarball %s", cycle.label, tarball_path.name)
+        logger.info("Downloading %s cycle %s", product.label, cycle.label)
+        files = self._download_cycle(product, cycle)
+        tarball_path = self._build_tarball(product, cycle, files)
         state = {
+            "product": product.key,
+            "product_label": product.label,
             "cycle": cycle.label,
             "files": [str(path.relative_to(self.data_root)) for path in files],
             "tarball": str(tarball_path.relative_to(self.data_root)),
@@ -85,104 +146,131 @@ class WaveDownloader:
     def current_state(self) -> Optional[Dict]:
         return self._read_state()
 
-    def _find_latest_cycle(self, max_days_back: int = 9) -> Optional[Cycle]:
+    def _find_latest_cycle(self, product: WaveProduct) -> Optional[Cycle]:
         now = datetime.now(timezone.utc)
-        candidate_days = [(now - timedelta(days=days)).strftime("%Y%m%d") for days in range(max_days_back + 1)]
+        candidate_days = [(now - timedelta(days=days)).strftime("%Y%m%d") for days in range(product.max_days_back + 1)]
         cycles = ["18", "12", "06", "00"]
 
         for day in candidate_days:
             for cycle in cycles:
-                logger.info("Probing cycle %s %sz", day, cycle)
-                if self._cycle_available(day, cycle):
-                    logger.info("Cycle %s %sz is available.", day, cycle)
+                logger.info("Probing %s cycle %s %sz", product.label, day, cycle)
+                if self._cycle_available(product, day, cycle):
+                    logger.info("Found %s cycle %s %sz.", product.label, day, cycle)
                     return Cycle(date=day, cycle=cycle)
-        logger.warning("No GEFS wave cycle found in the last %d days.", max_days_back)
+        logger.warning("No %s cycle found in the last %d days.", product.label, product.max_days_back)
         return None
 
-    def _cycle_available(self, day: str, cycle: str) -> bool:
-        filename = f"gefswave.t{cycle}z.c00.global.0p25.f000.grib2"
-        for base_url in BASE_URLS:
-            path = f"gefs.{day}/{cycle}/wave/gridded/{filename}"
-            url = self._build_url(base_url, path)
-            if not url:
-                logger.debug("Skipping %s for cycle %s %sz due to missing credentials.", base_url, day, cycle)
-                continue
-            try:
+    def _cycle_available(self, product: WaveProduct, day: str, cycle_hour: str) -> bool:
+        for endpoint in product.endpoints:
+            if self._probe_endpoint(product, endpoint, day, cycle_hour):
+                return True
+        return False
+
+    def _probe_endpoint(self, product: WaveProduct, endpoint: Endpoint, day: str, cycle_hour: str) -> bool:
+        url = self._build_url(product, endpoint, day, cycle_hour, hour=0)
+        if not url:
+            return False
+
+        try:
+            response = self.session.head(url, timeout=20)
+            if response.ok:
+                logger.info("Cycle %s %sz available via %s", day, cycle_hour, endpoint.label or endpoint.base_url)
+                return True
+
+            if response.status_code in (401, 403) and endpoint.needs_sas:
+                logger.warning(
+                    "Auth error probing %s on %s (HTTP %s). Refreshing credentials.",
+                    product.label,
+                    endpoint.label or endpoint.base_url,
+                    response.status_code,
+                )
+                self._invalidate_azure_token()
+                url = self._build_url(product, endpoint, day, cycle_hour, hour=0)
+                if not url:
+                    return False
                 response = self.session.head(url, timeout=20)
                 if response.ok:
-                    logger.info("Cycle %s %sz file found on %s", day, cycle, base_url)
                     return True
-                if response.status_code == 404:
-                    logger.debug("Cycle %s %sz missing on %s (404).", day, cycle, base_url)
-                elif response.status_code in (401, 403):
-                    logger.warning(
-                        "Cycle %s %sz request to %s returned %s; will retry after refreshing credentials.",
-                        day,
-                        cycle,
-                        base_url,
-                        response.status_code,
-                    )
-                    if base_url == AZURE_URL:
-                        self._invalidate_azure_token()
-                else:
-                    logger.debug(
-                        "Cycle %s %sz head request on %s returned HTTP %s.",
-                        day,
-                        cycle,
-                        base_url,
-                        response.status_code,
-                    )
-            except requests.RequestException as exc:
-                logger.debug("Error probing cycle %s %sz on %s: %s", day, cycle, base_url, exc)
+
+            if response.status_code not in (404, 400):
+                range_headers = {"Range": "bytes=0-0"}
+                with self.session.get(url, headers=range_headers, timeout=20, stream=True) as stream_resp:
+                    if stream_resp.status_code in (200, 206):
+                        logger.info(
+                            "Cycle %s %sz confirmed via range request on %s",
+                            day,
+                            cycle_hour,
+                            endpoint.label or endpoint.base_url,
+                        )
+                        return True
+        except requests.RequestException as exc:
+            logger.debug(
+                "Error probing %s cycle %s %sz on %s: %s",
+                product.label,
+                day,
+                cycle_hour,
+                endpoint.label or endpoint.base_url,
+                exc,
+            )
         return False
 
-    def _download_cycle(self, cycle: Cycle) -> List[Path]:
+    def _download_cycle(self, product: WaveProduct, cycle: Cycle) -> List[Path]:
         files: List[Path] = []
-        target_dir = self.data_root / cycle.date / cycle.cycle
+        target_dir = self.data_root / product.key / cycle.date / cycle.cycle
         target_dir.mkdir(parents=True, exist_ok=True)
 
-        for hour in FORECAST_HOURS:
-            filename = f"gefswave.t{cycle.cycle}z.c00.global.0p25.f{hour:03d}.grib2"
+        for hour in product.forecast_hours:
+            filename = product.filename_template.format(cycle=cycle.cycle, hour=hour)
             destination = target_dir / filename
             if destination.exists() and destination.stat().st_size > 0:
-                logger.debug("File %s already exists; skipping download.", destination)
+                logger.debug("File %s already exists; skipping.", destination)
                 files.append(destination)
                 continue
             logger.info("Downloading %s", filename)
-            self._download_with_fallback(cycle, filename, destination)
+            self._download_with_fallback(product, cycle, filename, destination)
             files.append(destination)
         return files
 
-    def _build_tarball(self, cycle: Cycle, files: Iterable[Path]) -> Path:
-        tarball_path = self.data_root / TARBALL_FILENAME
+    def _build_tarball(self, product: WaveProduct, cycle: Cycle, files: Iterable[Path]) -> Path:
+        tarball_path = self.data_root / f"{product.key}-latest.tar.gz"
         with tarfile.open(tarball_path, "w:gz") as tar:
             for file_path in files:
                 tar.add(file_path, arcname=file_path.relative_to(self.data_root))
         return tarball_path
 
-    def _download_with_fallback(self, cycle: Cycle, filename: str, destination: Path) -> None:
+    def _download_with_fallback(self, product: WaveProduct, cycle: Cycle, filename: str, destination: Path) -> None:
         errors = []
-        for base_url in BASE_URLS:
-            path = f"{cycle.directory}/wave/gridded/{filename}"
-            url = self._build_url(base_url, path)
-            if not url:
-                logger.debug("Skipping download from %s for %s due to missing URL.", base_url, filename)
-                continue
+        for endpoint in product.endpoints:
             try:
-                self._stream_to_file(url, destination, base_url)
+                self._stream_to_file(product, endpoint, cycle, filename, destination)
+                return
+            except RetryableAuthError:
+                logger.info("Retrying %s download on %s after refreshing credentials.", filename, endpoint.base_url)
+                self._stream_to_file(product, endpoint, cycle, filename, destination)
                 return
             except WaveDownloadError as exc:
                 errors.append(str(exc))
-                logger.warning("Failed to download %s from %s (%s).", filename, base_url, exc)
+                logger.warning("Failed to download %s from %s (%s).", filename, endpoint.base_url, exc)
         raise WaveDownloadError(f"All download attempts failed for {filename}: {'; '.join(errors)}")
 
-    def _stream_to_file(self, url: str, destination: Path, base_url: str) -> None:
+    def _stream_to_file(
+        self,
+        product: WaveProduct,
+        endpoint: Endpoint,
+        cycle: Cycle,
+        filename: str,
+        destination: Path,
+    ) -> None:
+        url = self._build_url(product, endpoint, cycle.date, cycle.cycle, filename=filename)
+        if not url:
+            raise WaveDownloadError(f"No accessible URL for endpoint {endpoint.base_url}")
+
         tmp_path = destination.with_suffix(destination.suffix + ".part")
         try:
             with self.session.get(url, stream=True, timeout=120) as response:
-                if response.status_code in (401, 403) and base_url == AZURE_URL:
+                if response.status_code in (401, 403) and endpoint.needs_sas:
                     self._invalidate_azure_token()
-                    response.raise_for_status()
+                    raise RetryableAuthError(f"Auth failure downloading {filename} from {endpoint.base_url}")
                 response.raise_for_status()
                 with tmp_path.open("wb") as handle:
                     for chunk in response.iter_content(chunk_size=1 << 20):
@@ -190,20 +278,44 @@ class WaveDownloader:
                             continue
                         handle.write(chunk)
             tmp_path.rename(destination)
+        except RetryableAuthError:
+            if tmp_path.exists():
+                tmp_path.unlink()
+            raise
         except requests.RequestException as exc:
             if tmp_path.exists():
                 tmp_path.unlink()
             logger.error("Failed to download %s: %s", url, exc)
             raise WaveDownloadError(f"Failed to download {url}: {exc}") from exc
 
-    def _build_url(self, base_url: str, path: str) -> Optional[str]:
-        if base_url == AZURE_URL:
+    def _build_url(
+        self,
+        product: WaveProduct,
+        endpoint: Endpoint,
+        day: str,
+        cycle_hour: str,
+        hour: Optional[int] = None,
+        filename: Optional[str] = None,
+    ) -> Optional[str]:
+        if filename is None:
+            if hour is None:
+                raise ValueError("Either hour or filename must be provided.")
+            filename = product.filename_template.format(cycle=cycle_hour, hour=hour)
+
+        base = endpoint.base_url.rstrip("/")
+        if endpoint.mode == "latest":
+            url = f"{base}/{filename}"
+        else:
+            directory = product.directory_template.format(date=day, cycle=cycle_hour)
+            url = f"{base}/{directory}/{filename}"
+
+        if endpoint.needs_sas:
             token = self._get_azure_sas_token()
             if not token:
                 return None
             token = token if token.startswith("?") else f"?{token}"
-            return f"{base_url}/{path}{token}"
-        return f"{base_url}/{path}"
+            return f"{url}{token}"
+        return url
 
     def _get_azure_sas_token(self) -> Optional[str]:
         if self._azure_sas_token:
@@ -238,11 +350,13 @@ class WaveDownloader:
         self._azure_sas_token = None
         self._azure_token_checked = False
 
-    def _cycle_complete(self, cycle: Cycle) -> bool:
-        target_dir = self.data_root / cycle.date / cycle.cycle
+    def _cycle_complete(self, product: WaveProduct, cycle: Cycle) -> bool:
+        target_dir = self.data_root / product.key / cycle.date / cycle.cycle
         if not target_dir.exists():
             return False
-        expected = {f"gefswave.t{cycle.cycle}z.c00.global.0p25.f{hour:03d}.grib2" for hour in FORECAST_HOURS}
+        expected = {
+            product.filename_template.format(cycle=cycle.cycle, hour=hour) for hour in product.forecast_hours
+        }
         existing = {path.name for path in target_dir.glob("*.grib2") if path.stat().st_size > 0}
         return expected.issubset(existing)
 
@@ -274,7 +388,7 @@ class WaveDownloadManager:
             if self._worker and self._worker.is_alive():
                 logger.info("Refresh already in progress; ignoring new trigger.")
                 return
-            logger.info("Starting GEFS wave refresh thread.")
+            logger.info("Starting wave refresh thread.")
             self._status = {"status": "running"}
             self._worker = threading.Thread(target=self._run_refresh, daemon=True)
             self._worker.start()
@@ -292,8 +406,8 @@ class WaveDownloadManager:
             result = self.downloader.ensure_latest_cycle()
             with self._lock:
                 self._status = {"status": "ready", "latest_state": result}
-            logger.info("GEFS wave refresh completed successfully.")
+            logger.info("Wave refresh completed successfully.")
         except Exception as exc:
             with self._lock:
                 self._status = {"status": "error", "message": str(exc)}
-            logger.exception("GEFS wave refresh failed: %s", exc)
+            logger.exception("Wave refresh failed: %s", exc)