Harden ISS and water publishers

Author: Sam-Bolling

publishers/iss/iss_publisher.py

@@ -27,8 +27,10 @@
 import math
 import os
 import sys
+import tempfile
 import time
 from datetime import datetime, timedelta, timezone
+from pathlib import Path
 from typing import Any
 from urllib.request import Request, urlopen
 
@@ -50,6 +52,25 @@
 NORAD_ID   = os.environ.get("NORAD_ID", "25544")
 ASSET_NAME = os.environ.get("ASSET_NAME", "ISS (ZARYA)")
 CELESTRAK_URL = f"https://celestrak.org/NORAD/elements/gp.php?CATNR={NORAD_ID}&FORMAT=JSON"
+TLE_CACHE_FILE = Path(os.environ.get(
+    "ISS_TLE_CACHE_FILE",
+    os.path.join(tempfile.gettempdir(), f"iss-{NORAD_ID}-omm.json"),
+))
+FALLBACK_OMM = {
+    "OBJECT_NAME": "ISS (ZARYA)",
+    "OBJECT_ID": "1998-067A",
+    "EPOCH": "2026-05-25T10:07:58.537056",
+    "MEAN_MOTION": 15.49365963,
+    "ECCENTRICITY": 0.00074701,
+    "INCLINATION": 51.633,
+    "RA_OF_ASC_NODE": 52.7989,
+    "ARG_OF_PERICENTER": 96.3597,
+    "MEAN_ANOMALY": 263.8242,
+    "NORAD_CAT_ID": 25544,
+    "BSTAR": 0.00018076148,
+    "MEAN_MOTION_DOT": 9.654e-5,
+    "MEAN_MOTION_DDOT": 0,
+}
 
 
 # ═══════════════════════════════════════════════════════════════════════════
@@ -63,18 +84,9 @@
 _tle_refresh_interval: float = 3600.0
 
 
-def fetch_tle_from_celestrak() -> Satrec:
+def _satrec_from_omm(omm: dict) -> Satrec:
     global _cached_satrec, _tle_fetched_at, _tle_epoch_str, _tle_epoch_dt
 
-    req = Request(CELESTRAK_URL, headers={"Accept": "application/json"})
-    with urlopen(req, timeout=30) as resp:
-        data = json.loads(resp.read().decode())
-
-    if isinstance(data, list) and len(data) > 0:
-        omm = data[0]
-    else:
-        raise RuntimeError(f"Unexpected CelesTrak response: {str(data)[:200]}")
-
     sat = Satrec()
     sat.sgp4init(
         WGS72, 'i',
@@ -102,6 +114,51 @@ def fetch_tle_from_celestrak() -> Satrec:
     return sat
 
 
+def _normalize_omm_response(data: Any) -> dict:
+    if isinstance(data, list) and len(data) > 0:
+        return data[0]
+    if isinstance(data, dict):
+        return data
+    raise RuntimeError(f"Unexpected CelesTrak response: {str(data)[:200]}")
+
+
+def _write_tle_cache(omm: dict) -> None:
+    try:
+        TLE_CACHE_FILE.parent.mkdir(parents=True, exist_ok=True)
+        TLE_CACHE_FILE.write_text(json.dumps(omm), encoding="utf-8")
+    except OSError as e:
+        print(f"  [WARN] Could not write TLE cache {TLE_CACHE_FILE}: {e}")
+
+
+def load_cached_or_fallback_tle(reason: str) -> Satrec:
+    print(f"  [WARN] TLE fetch failed ({reason}); using cached/fallback OMM")
+    if TLE_CACHE_FILE.exists():
+        try:
+            omm = _normalize_omm_response(json.loads(TLE_CACHE_FILE.read_text(encoding="utf-8")))
+            sat = _satrec_from_omm(omm)
+            print(f"  TLE cache epoch: {_tle_epoch_str}")
+            return sat
+        except Exception as e:
+            print(f"  [WARN] Could not load TLE cache {TLE_CACHE_FILE}: {e}")
+
+    sat = _satrec_from_omm(FALLBACK_OMM)
+    print(f"  TLE fallback epoch: {_tle_epoch_str}")
+    return sat
+
+
+def fetch_tle_from_celestrak() -> Satrec:
+    global _cached_satrec, _tle_fetched_at, _tle_epoch_str, _tle_epoch_dt
+
+    req = Request(CELESTRAK_URL, headers={"Accept": "application/json"})
+    with urlopen(req, timeout=30) as resp:
+        data = json.loads(resp.read().decode())
+
+    omm = _normalize_omm_response(data)
+    sat = _satrec_from_omm(omm)
+    _write_tle_cache(omm)
+    return sat
+
+
 def _epoch_to_jdsatepoch(epoch_str: str) -> float:
     dt = datetime.strptime(epoch_str, "%Y-%m-%dT%H:%M:%S.%f").replace(tzinfo=timezone.utc)
     jd, fr = _datetime_to_jd(dt)
@@ -111,7 +168,14 @@ def _epoch_to_jdsatepoch(epoch_str: str) -> float:
 def get_satrec() -> Satrec:
     global _cached_satrec, _tle_fetched_at
     if _cached_satrec is None or (time.time() - _tle_fetched_at) > _tle_refresh_interval:
-        fetch_tle_from_celestrak()
+        try:
+            fetch_tle_from_celestrak()
+        except Exception as e:
+            if _cached_satrec is None:
+                load_cached_or_fallback_tle(str(e))
+            else:
+                _tle_fetched_at = time.time()
+                print(f"  [WARN] TLE refresh failed ({e}); keeping epoch {_tle_epoch_str}")
     return _cached_satrec
 
 
@@ -227,8 +291,7 @@ def on_startup(self, args):
             fetch_tle_from_celestrak()
             print(f"  TLE epoch: {_tle_epoch_str}")
         except Exception as e:
-            print(f"  FATAL: Could not fetch TLE: {e}")
-            sys.exit(1)
+            load_cached_or_fallback_tle(str(e))
 
     def connect(self):
         """Connect to server. Uses REST mode when OSH_BASE_URL is set, SDK otherwise."""

publishers/usgs_water/usgs_water_publisher.py

@@ -30,6 +30,7 @@
 import json
 import os
 import random
+import re
 import ssl
 import sys
 import time
@@ -197,29 +198,63 @@ def __init__(self, station_filter: list[str] | None = None):
     def _system_uid(self, nwis_id: str) -> str:
         return f"urn:os4csapi:system:usgs-water:{nwis_id}:v1"
 
+    def _raw_datastream_ids(self, sys_id: str) -> dict[str, str]:
+        url = f"{self._base_url}/systems/{sys_id}/datastreams"
+        headers = {"Accept": "application/json", "Authorization": self._auth}
+        ctx = ssl.create_default_context()
+        ctx.check_hostname = False
+        ctx.verify_mode = ssl.CERT_NONE
+        with urlopen(Request(url, headers=headers), timeout=30, context=ctx) as resp:
+            text = resp.read().decode()
+
+        station_ds: dict[str, str] = {}
+        current_id = None
+        for key, value in re.findall(r'"(id|outputName)"\s*:\s*"([^"]+)"', text):
+            if key == "id":
+                current_id = value
+            elif key == "outputName" and current_id:
+                if value == DS_DISCHARGE_OUTPUT:
+                    station_ds["00060"] = current_id
+                elif value == DS_GAGE_HEIGHT_OUTPUT:
+                    station_ds["00065"] = current_id
+                current_id = None
+        return station_ds
+
     def connect(self):
         """Resolve system and datastream IDs for each station via REST API."""
         self._ds_ids.clear()
         connected = 0
         for st in self.stations:
             nwis_id = st["nwisId"]
             uid = self._system_uid(nwis_id)
-            sys_id = find_by_uid(self._base_url, self._auth, "systems", uid, no_cache=True)
-            if not sys_id:
-                print(f"  [WARN] System '{uid}' not found -- skipping {nwis_id}")
-                continue
-
-            # Find datastreams
-            ds_list = api_get(self._base_url, f"systems/{sys_id}/datastreams", self._auth)
+            sys_id = None
             station_ds = {}
-            if ds_list:
-                for item in ds_list.get("items", []):
-                    output_name = item.get("outputName", "")
-                    ds_id = item.get("id")
-                    if output_name == DS_DISCHARGE_OUTPUT and ds_id:
-                        station_ds["00060"] = ds_id
-                    elif output_name == DS_GAGE_HEIGHT_OUTPUT and ds_id:
-                        station_ds["00065"] = ds_id
+            try:
+                sys_id = find_by_uid(self._base_url, self._auth, "systems", uid, no_cache=True)
+                if not sys_id:
+                    print(f"  [WARN] System '{uid}' not found -- skipping {nwis_id}")
+                    continue
+
+                ds_list = api_get(self._base_url, f"systems/{sys_id}/datastreams", self._auth)
+                if ds_list:
+                    for item in ds_list.get("items", []):
+                        output_name = item.get("outputName", "")
+                        ds_id = item.get("id")
+                        if output_name == DS_DISCHARGE_OUTPUT and ds_id:
+                            station_ds["00060"] = ds_id
+                        elif output_name == DS_GAGE_HEIGHT_OUTPUT and ds_id:
+                            station_ds["00065"] = ds_id
+            except Exception as e:
+                if sys_id:
+                    station_ds = self._raw_datastream_ids(sys_id)
+                    if station_ds:
+                        print(f"  [WARN] Used raw datastream fallback for {nwis_id}: {e}")
+                    else:
+                        print(f"  [WARN] Could not resolve datastreams for {nwis_id}: {e}")
+                        continue
+                else:
+                    print(f"  [WARN] Could not resolve system for {nwis_id}: {e}")
+                    continue
 
             if not station_ds:
                 print(f"  [WARN] No datastreams found for {nwis_id} -- skipping")
@@ -231,6 +266,8 @@ def connect(self):
             print(f"  Connected: {nwis_id} -> sys={sys_id} ds=[{ds_summary}]")
 
         print(f"  Ready: {connected}/{len(self.stations)} stations connected")
+        if connected == 0:
+            raise RuntimeError("No USGS Water stations connected")
 
     def connect_with_retry(self, max_attempts=10, base_delay=5.0, max_delay=120.0):
         for attempt in range(1, max_attempts + 1):