#!/usr/bin/env python3 """ NWAL Scanner Weather V3 WeatherBug Spark Lightning Bridge v003 Purpose ------- Loads the WeatherBug Spark page in Playwright, captures its lightning JSONP response, extracts raw strike coordinates, keeps a rolling history, filters to a broad Southeast coverage box, and writes local JSON files for the broadcast server. Output files ------------ spark_lightning.json spark_lightning_status.json Install ------- python -m pip install playwright python -m playwright install chromium Run --- python spark_lightning_bridge_v003.py """ from __future__ import annotations import argparse import asyncio import json import math import os import signal import tempfile import time from dataclasses import dataclass from pathlib import Path from typing import Any from playwright.async_api import Browser, BrowserContext, Page, Response, async_playwright VERSION = "v003" DEFAULT_CENTER_LAT = 34.783003 DEFAULT_CENTER_LON = -87.671093 # Broad enough for the widest Southeast radar view, Gulf, and near-shore Atlantic. DEFAULT_BOUNDS = { "north": 39.5, "south": 24.0, "west": -98.0, "east": -74.0, } DEFAULT_HISTORY_MINUTES = 15 DEFAULT_STALE_SECONDS = 90 DEFAULT_RELOAD_SECONDS = 45 REFERENCE_LAT = 34.7998 REFERENCE_LON = -87.6773 NEW_STRIKE_WINDOW_SECONDS = 12 def load_location_profile(base_dir: Path) -> dict[str, Any]: try: profile_path = Path(os.environ.get('NWALS_LOCATION_PROFILE', str(base_dir / 'location_profile.json'))).resolve() value = json.loads(profile_path.read_text(encoding="utf-8")) return value if isinstance(value, dict) else {} except Exception: return {} # Windows can briefly lock a JSON file while the local server, antivirus, # indexer, or another process is reading it. Retry atomic replacements rather # than allowing one transient lock to terminate the bridge. WRITE_RETRY_ATTEMPTS = 10 WRITE_RETRY_DELAY_SECONDS = 0.20 WRITE_WARNING_AFTER_SECONDS = 30 SPARK_PAGE_TEMPLATE = ( "https://lxapp.weatherbug.net/v2/lxapp_impl.html" "?lat={lat:.6f}&lon={lon:.6f}&v=1.2.0" ) SPARK_URL_MARKERS = ( "/api/lx/lightning/v2/spark", "lightning/v2/spark", ) @dataclass(frozen=True) class Strike: lat: float lon: float timestamp: int # Unix seconds @property def id(self) -> str: # Timestamp plus 5-decimal coordinates gives a stable identity while # preserving the source precision seen in Spark responses. return f"{self.lat:.5f}:{self.lon:.5f}:{self.timestamp}" def as_dict(self, now_s: int) -> dict[str, Any]: return { "id": self.id, "lat": self.lat, "lon": self.lon, "timestamp": self.timestamp, "timestamp_ms": self.timestamp * 1000, "age_seconds": max(0, now_s - self.timestamp), "provider": "weatherbug_spark", } def haversine_miles(lat1: float, lon1: float, lat2: float, lon2: float) -> float: radius_miles = 3958.7613 phi1 = math.radians(lat1) phi2 = math.radians(lat2) dphi = math.radians(lat2 - lat1) dlambda = math.radians(lon2 - lon1) a = ( math.sin(dphi / 2) ** 2 + math.cos(phi1) * math.cos(phi2) * math.sin(dlambda / 2) ** 2 ) return radius_miles * 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a)) def bearing_degrees(lat1: float, lon1: float, lat2: float, lon2: float) -> float: phi1 = math.radians(lat1) phi2 = math.radians(lat2) dlambda = math.radians(lon2 - lon1) y = math.sin(dlambda) * math.cos(phi2) x = ( math.cos(phi1) * math.sin(phi2) - math.sin(phi1) * math.cos(phi2) * math.cos(dlambda) ) return (math.degrees(math.atan2(y, x)) + 360) % 360 def compass_direction(degrees: float) -> str: labels = [ "N", "NNE", "NE", "ENE", "E", "ESE", "SE", "SSE", "S", "SSW", "SW", "WSW", "W", "WNW", "NW", "NNW", ] return labels[int((degrees + 11.25) // 22.5) % 16] class SparkBridge: def __init__( self, output_dir: Path, center_lat: float, center_lon: float, bounds: dict[str, float], history_minutes: int, stale_seconds: int, reload_seconds: int, headless: bool, reference_name: str, reference_lat: float, reference_lon: float, monitoring_radii: list[float], ) -> None: self.output_dir = output_dir self.center_lat = center_lat self.center_lon = center_lon self.bounds = bounds self.history_seconds = history_minutes * 60 self.stale_seconds = stale_seconds self.reload_seconds = reload_seconds self.headless = headless self.reference_name = reference_name self.reference_lat = reference_lat self.reference_lon = reference_lon self.monitoring_radii = monitoring_radii self.strikes: dict[str, Strike] = {} self.last_capture_wall_time: float | None = None self.last_source_timestamp: int | None = None self.last_error: str | None = None self.capture_count = 0 self.total_source_records = 0 self.total_in_bounds_records = 0 self.running = True self.previous_published_ids: set[str] = set() self._write_lock = asyncio.Lock() self.write_failure_started_at: float | None = None self.write_failure_count = 0 self.last_write_error: str | None = None self.last_successful_write_at: float | None = None self.data_path = output_dir / "spark_lightning.json" self.status_path = output_dir / "spark_lightning_status.json" def in_bounds(self, lat: float, lon: float) -> bool: return ( self.bounds["south"] <= lat <= self.bounds["north"] and self.bounds["west"] <= lon <= self.bounds["east"] ) @staticmethod def parse_jsonp(text: str) -> dict[str, Any]: text = text.strip() start = text.find("(") end = text.rfind(")") if start < 0 or end <= start: raise ValueError("Spark response was not valid JSONP") payload = text[start + 1 : end].strip() return json.loads(payload) @staticmethod def _write_temp_json(path: Path, payload: Any) -> str: """Write and fsync a temporary JSON file beside the destination.""" path.parent.mkdir(parents=True, exist_ok=True) fd, tmp_name = tempfile.mkstemp( prefix=path.name + ".", suffix=".tmp", dir=str(path.parent), ) try: with os.fdopen(fd, "w", encoding="utf-8") as handle: json.dump(payload, handle, indent=2, separators=(",", ": ")) handle.write("\n") handle.flush() os.fsync(handle.fileno()) return tmp_name except Exception: try: os.close(fd) except OSError: pass if os.path.exists(tmp_name): try: os.unlink(tmp_name) except OSError: pass raise async def atomic_json_write(self, path: Path, payload: Any) -> bool: """ Atomically replace a JSON output file. Windows may temporarily deny os.replace() while another process has the destination open. Retry that operation and return False instead of raising after all attempts. The caller can then continue running and publish again on the next capture. """ tmp_name: str | None = None try: tmp_name = self._write_temp_json(path, payload) for attempt in range(1, WRITE_RETRY_ATTEMPTS + 1): try: os.replace(tmp_name, path) tmp_name = None return True except PermissionError as exc: if attempt >= WRITE_RETRY_ATTEMPTS: self.last_write_error = ( f"{type(exc).__name__}: {exc}" ) return False await asyncio.sleep(WRITE_RETRY_DELAY_SECONDS) except OSError as exc: # WinError 5 can occasionally arrive as a generic OSError. if getattr(exc, "winerror", None) == 5: if attempt >= WRITE_RETRY_ATTEMPTS: self.last_write_error = ( f"{type(exc).__name__}: {exc}" ) return False await asyncio.sleep(WRITE_RETRY_DELAY_SECONDS) continue raise return False except Exception as exc: self.last_write_error = f"{type(exc).__name__}: {exc}" return False finally: if tmp_name and os.path.exists(tmp_name): try: os.unlink(tmp_name) except OSError: pass def prune_history(self, now_s: int) -> None: cutoff = now_s - self.history_seconds stale_ids = [ strike_id for strike_id, strike in self.strikes.items() if strike.timestamp < cutoff ] for strike_id in stale_ids: self.strikes.pop(strike_id, None) async def write_outputs(self) -> bool: async with self._write_lock: now_s = int(time.time()) self.prune_history(now_s) sorted_strikes = sorted( self.strikes.values(), key=lambda strike: strike.timestamp, reverse=True, ) current_ids = {strike.id for strike in sorted_strikes} new_ids = current_ids - self.previous_published_ids new_strikes = [ strike for strike in sorted_strikes if strike.id in new_ids and now_s - strike.timestamp <= NEW_STRIKE_WINDOW_SECONDS ] distance_rows = [] for strike in sorted_strikes: miles = haversine_miles( self.reference_lat, self.reference_lon, strike.lat, strike.lon, ) bearing = bearing_degrees( self.reference_lat, self.reference_lon, strike.lat, strike.lon, ) distance_rows.append((miles, bearing, strike)) nearest = None if distance_rows: miles, bearing, strike = min(distance_rows, key=lambda row: row[0]) nearest = { "distance_miles": round(miles, 1), "bearing_degrees": round(bearing), "direction": compass_direction(bearing), "strike_id": strike.id, "timestamp": strike.timestamp, "age_seconds": max(0, now_s - strike.timestamp), } radius_counts = { f"within_{int(radius) if float(radius).is_integer() else radius:g}_miles": sum(1 for miles, _, _ in distance_rows if miles <= radius) for radius in self.monitoring_radii } capture_age = ( None if self.last_capture_wall_time is None else round(time.time() - self.last_capture_wall_time, 1) ) healthy = ( self.last_capture_wall_time is not None and capture_age is not None and capture_age <= self.stale_seconds ) data = { "provider": "weatherbug_spark", "bridge_version": VERSION, "generated_at": now_s, "generated_at_ms": now_s * 1000, "history_minutes": self.history_seconds // 60, "coverage": self.bounds, "reference_point": { "name": self.reference_name, "lat": self.reference_lat, "lon": self.reference_lon, }, "strike_count": len(sorted_strikes), "new_strike_count": len(new_strikes), "latest_strike_timestamp": self.last_source_timestamp, "nearest_strike": nearest, "radius_counts": radius_counts, "new_strikes": [strike.as_dict(now_s) for strike in new_strikes], "strikes": [strike.as_dict(now_s) for strike in sorted_strikes], } status = { "provider": "weatherbug_spark", "bridge_version": VERSION, "status": "healthy" if healthy else "stale_or_starting", "healthy": healthy, "capture_age_seconds": capture_age, "stale_after_seconds": self.stale_seconds, "last_capture_epoch": ( None if self.last_capture_wall_time is None else int(self.last_capture_wall_time) ), "latest_strike_timestamp": self.last_source_timestamp, "strike_count": len(sorted_strikes), "new_strike_count": len(new_strikes), "capture_count": self.capture_count, "last_source_record_count": self.total_source_records, "last_in_bounds_record_count": self.total_in_bounds_records, "coverage": self.bounds, "reference_point": data["reference_point"], "nearest_strike": nearest, "radius_counts": radius_counts, "last_error": self.last_error, "last_write_error": self.last_write_error, "write_failure_count": self.write_failure_count, "last_successful_write_epoch": ( None if self.last_successful_write_at is None else int(self.last_successful_write_at) ), } data_ok = await self.atomic_json_write(self.data_path, data) status_ok = await self.atomic_json_write(self.status_path, status) writes_ok = data_ok and status_ok if writes_ok: if self.write_failure_started_at is not None: outage_seconds = round( time.time() - self.write_failure_started_at, 1, ) print( f"[SPARK] output writes recovered after " f"{outage_seconds}s" ) self.write_failure_started_at = None self.write_failure_count = 0 self.last_write_error = None self.last_successful_write_at = time.time() self.previous_published_ids = current_ids return True self.write_failure_count += 1 if self.write_failure_started_at is None: self.write_failure_started_at = time.time() failure_age = time.time() - self.write_failure_started_at print( f"[SPARK] output write busy; keeping bridge alive " f"(failure {self.write_failure_count}, " f"age {failure_age:.1f}s): {self.last_write_error}" ) if failure_age >= WRITE_WARNING_AFTER_SECONDS: print( "[SPARK] WARNING: output files have remained locked " f"for {failure_age:.1f}s; capture will continue and " "publishing will retry automatically" ) return False async def process_response(self, response: Response) -> None: url_lower = response.url.lower() if not any(marker in url_lower for marker in SPARK_URL_MARKERS): return try: text = await response.text() payload = self.parse_jsonp(text) records = (((payload or {}).get("r") or {}).get("plg") or []) # Spark sometimes emits an empty bootstrap/cleanup response during # page load. It is not a meaningful feed update, so do not let it # refresh health timestamps or inflate capture counts. if not records: print("[SPARK] ignored empty response") return now_s = int(time.time()) accepted = 0 latest_ts = self.last_source_timestamp for record in records: try: lat = float(record["la"]) lon = float(record["lo"]) timestamp = int(record["t"]) except (KeyError, TypeError, ValueError): continue if not ( math.isfinite(lat) and math.isfinite(lon) and -90 <= lat <= 90 and -180 <= lon <= 180 ): continue if not self.in_bounds(lat, lon): continue # Ignore absurdly future or ancient values. A small future # allowance protects against clock skew. if timestamp > now_s + 300: continue if timestamp < now_s - self.history_seconds - 120: continue strike = Strike(lat=lat, lon=lon, timestamp=timestamp) self.strikes[strike.id] = strike accepted += 1 latest_ts = timestamp if latest_ts is None else max(latest_ts, timestamp) self.last_capture_wall_time = time.time() self.last_source_timestamp = latest_ts self.last_error = None self.capture_count += 1 self.total_source_records = len(records) self.total_in_bounds_records = accepted await self.write_outputs() print( f"[SPARK] capture={self.capture_count} " f"source={len(records)} southeast={accepted} " f"rolling={len(self.strikes)}" ) except Exception as exc: self.last_error = f"{type(exc).__name__}: {exc}" try: await self.write_outputs() except Exception as write_exc: # Last-resort containment. The bridge must stay alive even if # both response processing and status publishing fail. self.last_write_error = ( f"{type(write_exc).__name__}: {write_exc}" ) print(f"[SPARK] response error: {self.last_error}") async def attach_page(self, page: Page) -> None: page.on( "response", lambda response: asyncio.create_task(self.process_response(response)), ) async def keep_page_alive(self, page: Page) -> None: url = SPARK_PAGE_TEMPLATE.format( lat=self.center_lat, lon=self.center_lon, ) while self.running: try: print(f"[SPARK] loading {url}") await page.goto(url, wait_until="domcontentloaded", timeout=60_000) started = time.time() while self.running and time.time() - started < self.reload_seconds: await asyncio.sleep(2) await self.write_outputs() if ( self.last_capture_wall_time is not None and time.time() - self.last_capture_wall_time > self.stale_seconds ): print("[SPARK] feed stale; reloading page") break except Exception as exc: self.last_error = f"{type(exc).__name__}: {exc}" try: await self.write_outputs() except Exception as write_exc: self.last_write_error = ( f"{type(write_exc).__name__}: {write_exc}" ) print(f"[SPARK] page error: {self.last_error}") await asyncio.sleep(5) async def run(self) -> None: self.output_dir.mkdir(parents=True, exist_ok=True) await self.write_outputs() async with async_playwright() as playwright: browser: Browser = await playwright.chromium.launch( headless=self.headless, args=[ "--disable-background-timer-throttling", "--disable-backgrounding-occluded-windows", "--disable-renderer-backgrounding", ], ) context: BrowserContext = await browser.new_context( viewport={"width": 1280, "height": 720}, locale="en-US", ) page = await context.new_page() await self.attach_page(page) try: await self.keep_page_alive(page) finally: await context.close() await browser.close() def stop(self) -> None: self.running = False def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description="Capture and normalize WeatherBug Spark lightning data." ) parser.add_argument( "--output-dir", default=".", help="Directory for Spark JSON output files. Default: current directory", ) parser.add_argument("--lat", type=float, default=DEFAULT_CENTER_LAT) parser.add_argument("--lon", type=float, default=DEFAULT_CENTER_LON) parser.add_argument( "--history-minutes", type=int, default=DEFAULT_HISTORY_MINUTES, ) parser.add_argument( "--stale-seconds", type=int, default=DEFAULT_STALE_SECONDS, ) parser.add_argument( "--reload-seconds", type=int, default=DEFAULT_RELOAD_SECONDS, ) parser.add_argument( "--show-browser", action="store_true", help="Show Chromium for troubleshooting.", ) parser.add_argument("--north", type=float, default=DEFAULT_BOUNDS["north"]) parser.add_argument("--south", type=float, default=DEFAULT_BOUNDS["south"]) parser.add_argument("--west", type=float, default=DEFAULT_BOUNDS["west"]) parser.add_argument("--east", type=float, default=DEFAULT_BOUNDS["east"]) return parser async def async_main() -> int: args = build_parser().parse_args() profile = load_location_profile(Path(__file__).resolve().parent) lightning = profile.get("lightning", {}) if isinstance(profile.get("lightning"), dict) else {} reference = lightning.get("reference", {}) if isinstance(lightning.get("reference"), dict) else {} capture_center = lightning.get("captureCenter", {}) if isinstance(lightning.get("captureCenter"), dict) else {} coverage = lightning.get("coverage", {}) if isinstance(lightning.get("coverage"), dict) else {} radii = lightning.get("monitoringRadiiMiles", [10, 25, 50, 100]) radii = [float(value) for value in radii if float(value) > 0] if isinstance(radii, list) else [10.0, 25.0, 50.0, 100.0] if not radii: radii = [10.0, 25.0, 50.0, 100.0] # Command-line values still win when explicitly changed; profile supplies normal launcher defaults. if args.lat == DEFAULT_CENTER_LAT: args.lat = float(capture_center.get("latitude", args.lat)) if args.lon == DEFAULT_CENTER_LON: args.lon = float(capture_center.get("longitude", args.lon)) for key in ("north", "south", "west", "east"): if getattr(args, key) == DEFAULT_BOUNDS[key]: setattr(args, key, float(coverage.get(key, getattr(args, key)))) if args.history_minutes < 1: raise SystemExit("--history-minutes must be at least 1") if args.south >= args.north: raise SystemExit("--south must be less than --north") if args.west >= args.east: raise SystemExit("--west must be less than --east") bridge = SparkBridge( output_dir=Path(args.output_dir).resolve(), center_lat=args.lat, center_lon=args.lon, bounds={ "north": args.north, "south": args.south, "west": args.west, "east": args.east, }, history_minutes=args.history_minutes, stale_seconds=args.stale_seconds, reload_seconds=args.reload_seconds, headless=not args.show_browser, reference_name=str(lightning.get("referenceName", "Florence, Alabama")), reference_lat=float(reference.get("latitude", REFERENCE_LAT)), reference_lon=float(reference.get("longitude", REFERENCE_LON)), monitoring_radii=radii, ) loop = asyncio.get_running_loop() for signal_name in ("SIGINT", "SIGTERM"): sig = getattr(signal, signal_name, None) if sig is not None: try: loop.add_signal_handler(sig, bridge.stop) except NotImplementedError: pass await bridge.run() return 0 def main() -> int: try: return asyncio.run(async_main()) except KeyboardInterrupt: return 0 if __name__ == "__main__": raise SystemExit(main())