Add mirror fallback and log viewer

app.py

@@ -1,15 +1,42 @@
+import logging
 import os
+from collections import deque
 from pathlib import Path
 from threading import Event
 from typing import Dict, Optional, Tuple
 
 import gradio as gr
 
-from gefs_wave import WaveDownloadManager
+from gefs_wave import WaveDownloadManager, logger as wave_logger
 
 DATA_ROOT = Path(os.environ.get("GEFS_WAVE_DATA_DIR", "data/gefswave"))
 manager = WaveDownloadManager(DATA_ROOT)
 _refresh_started = Event()
+_log_buffer: deque[str] = deque(maxlen=200)
+
+
+class UILogHandler(logging.Handler):
+    def emit(self, record) -> None:  # type: ignore[override]
+        try:
+            message = self.format(record)
+        except Exception:
+            message = record.getMessage()
+        _log_buffer.append(message)
+
+
+log_formatter = logging.Formatter("%Y-%m-%d %H:%M:%S %(levelname)s %(message)s")
+ui_log_handler = UILogHandler()
+ui_log_handler.setFormatter(log_formatter)
+
+if not any(isinstance(handler, UILogHandler) for handler in wave_logger.handlers):
+    wave_logger.addHandler(ui_log_handler)
+wave_logger.setLevel(logging.INFO)
+
+
+def _log_text() -> str:
+    if not _log_buffer:
+        return "No log messages yet."
+    return "\n".join(_log_buffer)
 
 
 def ensure_refresh() -> None:
@@ -45,22 +72,22 @@ def format_status(status: Dict, notice: Optional[str] = None) -> Tuple[Dict, str
     return status, "\n".join(lines)
 
 
-def get_status() -> Tuple[Dict, str, Optional[str]]:
+def get_status() -> Tuple[Dict, str, Optional[str], str]:
     ensure_refresh()
     status = manager.status()
     status_dict, status_text = format_status(status)
-    return status_dict, status_text, None
+    return status_dict, status_text, None, _log_text()
 
 
-def trigger_refresh() -> Tuple[Dict, str, Optional[str]]:
+def trigger_refresh() -> Tuple[Dict, str, Optional[str], str]:
     ensure_refresh()
     manager.trigger_refresh()
     status = manager.status()
     status_dict, status_text = format_status(status)
-    return status_dict, status_text, None
+    return status_dict, status_text, None, _log_text()
 
 
-def download_tarball() -> Tuple[Dict, str, Optional[str]]:
+def download_tarball() -> Tuple[Dict, str, Optional[str], str]:
     ensure_refresh()
     status = manager.status()
     state = status.get("latest_state", {})
@@ -70,17 +97,17 @@ def download_tarball() -> Tuple[Dict, str, Optional[str]]:
         status_dict, status_text = format_status(
             status, "Wave dataset not available yet. Trigger a refresh and try again."
         )
-        return status_dict, status_text, None
+        return status_dict, status_text, None, _log_text()
 
     tarball_path = DATA_ROOT / tarball_rel
     if not tarball_path.exists():
         status_dict, status_text = format_status(
             status, "Tarball missing on disk. Trigger a refresh and try again."
         )
-        return status_dict, status_text, None
+        return status_dict, status_text, None, _log_text()
 
     status_dict, status_text = format_status(status)
-    return status_dict, status_text, str(tarball_path)
+    return status_dict, status_text, str(tarball_path), _log_text()
 
 
 with gr.Blocks(title="GFES Wave 0 Downloader") as demo:
@@ -96,14 +123,15 @@ with gr.Blocks(title="GFES Wave 0 Downloader") as demo:
     )
     status_json = gr.JSON(label="Downloader Status")
     status_md = gr.Markdown()
+    logs_box = gr.Textbox(label="Recent Logs", lines=12, max_lines=12, interactive=False)
 
     with gr.Row():
         refresh_button = gr.Button("Trigger Refresh", variant="secondary")
         download_button = gr.DownloadButton("Download Latest Tarball", variant="primary")
 
-    demo.load(get_status, outputs=[status_json, status_md, download_button])
-    refresh_button.click(trigger_refresh, outputs=[status_json, status_md, download_button])
-    download_button.click(download_tarball, outputs=[status_json, status_md, download_button])
+    demo.load(get_status, outputs=[status_json, status_md, download_button, logs_box])
+    refresh_button.click(trigger_refresh, outputs=[status_json, status_md, download_button, logs_box])
+    download_button.click(download_tarball, outputs=[status_json, status_md, download_button, logs_box])
 
 demo.queue()
 

gefs_wave.py

@@ -11,7 +11,9 @@ from typing import Dict, Iterable, List, Optional
 import requests
 
 
-BASE_URL = "https://nomads.ncep.noaa.gov/pub/data/nccf/com/gens/prod"
+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]
 FORECAST_HOURS: List[int] = list(range(0, 387, 3))
 STATE_FILENAME = "state.json"
 TARBALL_FILENAME = "gefswave-wave0-latest.tar.gz"
@@ -80,7 +82,7 @@ class WaveDownloader:
     def current_state(self) -> Optional[Dict]:
         return self._read_state()
 
-    def _find_latest_cycle(self, max_days_back: int = 5) -> Optional[Cycle]:
+    def _find_latest_cycle(self, max_days_back: int = 9) -> 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)]
         cycles = ["18", "12", "06", "00"]
@@ -95,17 +97,26 @@ class WaveDownloader:
         return None
 
     def _cycle_available(self, day: str, cycle: str) -> bool:
-        url = f"{BASE_URL}/gefs.{day}/{cycle}/wave/gridded/"
-        try:
-            response = self.session.get(url, timeout=20)
-            if not response.ok:
-                logger.debug("Cycle %s %sz check failed with HTTP %s.", day, cycle, response.status_code)
-                return False
-            pattern = re.compile(rf"gefswave\.t{cycle}z\.c00\.global\.0p25\.f000\.grib2")
-            return bool(pattern.search(response.text))
-        except requests.RequestException as exc:
-            logger.debug("Error probing cycle %s %sz: %s", day, cycle, exc)
-            return False
+        pattern = re.compile(rf"gefswave\.t{cycle}z\.c00\.global\.0p25\.f000\.grib2")
+        for base_url in BASE_URLS:
+            url = f"{base_url}/gefs.{day}/{cycle}/wave/gridded/"
+            try:
+                response = self.session.get(url, timeout=20)
+                if not response.ok:
+                    logger.debug(
+                        "Cycle %s %sz check failed on %s with HTTP %s.",
+                        day,
+                        cycle,
+                        base_url,
+                        response.status_code,
+                    )
+                    continue
+                if pattern.search(response.text):
+                    logger.info("Cycle %s %sz present on %s", day, cycle, base_url)
+                    return True
+            except requests.RequestException as exc:
+                logger.debug("Error probing cycle %s %sz on %s: %s", day, cycle, base_url, exc)
+        return False
 
     def _download_cycle(self, cycle: Cycle) -> List[Path]:
         files: List[Path] = []
@@ -114,14 +125,13 @@ class WaveDownloader:
 
         for hour in FORECAST_HOURS:
             filename = f"gefswave.t{cycle.cycle}z.c00.global.0p25.f{hour:03d}.grib2"
-            url = f"{BASE_URL}/{cycle.directory}/wave/gridded/{filename}"
             destination = target_dir / filename
             if destination.exists() and destination.stat().st_size > 0:
                 logger.debug("File %s already exists; skipping download.", destination)
                 files.append(destination)
                 continue
             logger.info("Downloading %s", filename)
-            self._stream_to_file(url, destination)
+            self._download_with_fallback(cycle, filename, destination)
             files.append(destination)
         return files
 
@@ -132,6 +142,18 @@ class WaveDownloader:
                 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:
+        errors = []
+        for base_url in BASE_URLS:
+            url = f"{base_url}/{cycle.directory}/wave/gridded/{filename}"
+            try:
+                self._stream_to_file(url, destination)
+                return
+            except WaveDownloadError as exc:
+                errors.append(str(exc))
+                logger.warning("Failed to download %s from %s (%s).", filename, base_url, exc)
+        raise WaveDownloadError(f"All download attempts failed for {filename}: {'; '.join(errors)}")
+
     def _stream_to_file(self, url: str, destination: Path) -> None:
         tmp_path = destination.with_suffix(destination.suffix + ".part")
         try: