#!/usr/bin/env python3 """ NW AL Scanner Baron Lightning Bridge v003 ----------------------------------------- Uses a dedicated persistent Google Chrome profile so the authorized Baron widget remembers that the Lightning layer is enabled. FIRST RUN --------- 1. Run this script normally. 2. A regular Chrome window opens with the Baron map. 3. Open the map's layer/product menu and enable Lightning once. 4. Leave the window open. The bridge captures the Shoals-region lightning requests and writes baron_lightning.json. LATER RUNS ---------- The same Chrome profile is reused, so Lightning should already be enabled. Optional: Set BARON_HEADLESS=1 after the first successful visible run. Requirement: py -m pip install playwright A separate Playwright Chromium download is not required because this script uses the installed Google Chrome channel. """ import asyncio import json import os import time from datetime import datetime, timezone from pathlib import Path from urllib.parse import parse_qs, urlparse from playwright.async_api import async_playwright, TimeoutError as PlaywrightTimeoutError BASE_DIR = Path(__file__).resolve().parent PROFILE_DIR = BASE_DIR / "baron_chrome_profile" OUTPUT_FILE = BASE_DIR / "baron_lightning.json" STATUS_FILE = BASE_DIR / "baron_lightning_status.json" WIDGET_URL = ( "https://staticbaronwebapps.velocityweather.com/" "digitial_wx/widgets/mapv2/index.html" "?initjson=/digitial_wx/widgets/dcms/" "017da136-5149-4082-aed2-93cd4f1a1683/live/init.json" "#8.58/34.774/-87.6187" ) SHOALS_LAT = 34.80 SHOALS_LON = -87.68 RELOAD_EVERY_SECONDS = 25 * 60 STALE_AFTER_SECONDS = 5 * 60 INITIAL_CAPTURE_WAIT_SECONDS = 120 HEADLESS = os.environ.get("BARON_HEADLESS", "").strip().lower() in { "1", "true", "yes", "on" } def utc_now_iso(): return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") def write_json_atomic(path: Path, payload): temp = path.with_suffix(path.suffix + ".tmp") temp.write_text(json.dumps(payload, indent=2), encoding="utf-8") temp.replace(path) def write_status(state, **extra): write_json_atomic( STATUS_FILE, { "provider": "baron", "state": state, "updated": utc_now_iso(), "profile_dir": str(PROFILE_DIR), "headless": HEADLESS, **extra, }, ) def extract_region_bounds(url: str): if "/reports/lightning/region.json" not in url: return None try: query = parse_qs(urlparse(url).query) return { "north": float(query["n_lat"][0]), "south": float(query["s_lat"][0]), "west": float(query["w_lon"][0]), "east": float(query["e_lon"][0]), } except (KeyError, IndexError, TypeError, ValueError): return None def region_contains_shoals(url: str): bounds = extract_region_bounds(url) if not bounds: return False return ( bounds["south"] <= SHOALS_LAT <= bounds["north"] and bounds["west"] <= SHOALS_LON <= bounds["east"] ) def normalize_payload(payload): # Current Baron response shape: # {"lightning": {"meta": {...}, "data": [...]}} if isinstance(payload, dict): lightning = payload.get("lightning") if isinstance(lightning, dict): payload = lightning.get("data", []) elif isinstance(payload.get("data"), list): payload = payload["data"] if not isinstance(payload, list): return [] now_ms = int(time.time() * 1000) strikes = [] seen = set() for item in payload: if not isinstance(item, dict): continue coords = ( item.get("coordinates") or item.get("coord") or item.get("location") ) if not isinstance(coords, (list, tuple)) or len(coords) < 2: continue try: lon = float(coords[0]) lat = float(coords[1]) except (TypeError, ValueError): continue raw_time = item.get("time") or item.get("timestamp") or item.get("t") ts_ms = None if isinstance(raw_time, (int, float)): ts_ms = int(raw_time) if ts_ms < 1_000_000_000_000: ts_ms *= 1000 elif isinstance(raw_time, str): try: ts_ms = int( datetime.fromisoformat( raw_time.replace("Z", "+00:00") ).timestamp() * 1000 ) except ValueError: continue if ts_ms is None: continue # Preserve enough history for the V3 15-minute display window. if now_ms - ts_ms > 20 * 60 * 1000: continue if ts_ms - now_ms > 5 * 60 * 1000: continue key = (round(lat, 4), round(lon, 4), round(ts_ms / 1000)) if key in seen: continue seen.add(key) strikes.append( { "lat": lat, "lon": lon, "t": ts_ms, "type": item.get("type", ""), "intensity": item.get("intensity"), "multiplicity": item.get("multiplicity"), } ) strikes.sort(key=lambda strike: strike["t"]) return strikes async def run_bridge(): PROFILE_DIR.mkdir(exist_ok=True) last_capture = 0.0 latest_count = 0 captured_once = False capture_event = asyncio.Event() async with async_playwright() as playwright: print( f"[BARON] Opening persistent Chrome profile: {PROFILE_DIR}", flush=True, ) print( f"[BARON] Mode: {'headless' if HEADLESS else 'visible'}", flush=True, ) context = await playwright.chromium.launch_persistent_context( user_data_dir=str(PROFILE_DIR), channel="chrome", headless=HEADLESS, viewport={"width": 1440, "height": 900}, locale="en-US", args=[ "--no-first-run", "--no-default-browser-check", "--disable-background-timer-throttling", "--disable-renderer-backgrounding", ], ) pages = context.pages page = pages[0] if pages else await context.new_page() async def handle_response(response): nonlocal last_capture, latest_count, captured_once if not region_contains_shoals(response.url): return bounds = extract_region_bounds(response.url) print( "[BARON] Found Shoals lightning region " f"{bounds['south']},{bounds['west']} to " f"{bounds['north']},{bounds['east']}", flush=True, ) try: payload = await response.json() strikes = normalize_payload(payload) write_json_atomic(OUTPUT_FILE, strikes) last_capture = time.time() latest_count = len(strikes) captured_once = True capture_event.set() write_status( "live", strike_count=latest_count, age_seconds=0, region=bounds, ) print( f"[BARON] Captured Shoals-region lightning: " f"{latest_count} recent strikes", flush=True, ) except Exception as exc: write_status("error", error=str(exc)) print( f"[BARON] Could not parse lightning response: {exc}", flush=True, ) page.on("response", handle_response) while True: try: capture_event.clear() write_status( "loading", strike_count=latest_count, instruction=( "Enable Lightning once in the visible Baron map " "if it is not already enabled." ), ) print("[BARON] Loading authorized Baron widget...", flush=True) await page.goto( WIDGET_URL, wait_until="domcontentloaded", timeout=60_000, ) await page.bring_to_front() if not HEADLESS: print("", flush=True) print( "[BARON] FIRST-RUN CHECK: In the Chrome window, " "open the Baron layers/products menu and enable Lightning.", flush=True, ) print( "[BARON] Baron should remember this setting in the " "dedicated profile for future launches.", flush=True, ) print("", flush=True) try: await asyncio.wait_for( capture_event.wait(), timeout=INITIAL_CAPTURE_WAIT_SECONDS, ) except asyncio.TimeoutError: if not captured_once: write_status( "waiting_for_lightning_layer", strike_count=latest_count, instruction=( "Open the Baron layer menu in the visible Chrome " "window and enable Lightning once." ), ) print( "[BARON] No Shoals lightning request captured yet.", flush=True, ) print( "[BARON] Enable Lightning in the visible Baron map. " "The bridge will capture it automatically.", flush=True, ) else: print( "[BARON] No new request during initial wait; " "continuing with the saved profile.", flush=True, ) cycle_started = time.time() while time.time() - cycle_started < RELOAD_EVERY_SECONDS: if last_capture: age = time.time() - last_capture if age > STALE_AFTER_SECONDS: write_status( "stale", strike_count=latest_count, age_seconds=int(age), ) await page.wait_for_timeout(15_000) print( "[BARON] Reloading widget for a fresh authorization window.", flush=True, ) except PlaywrightTimeoutError as exc: write_status("error", error=f"Widget timeout: {exc}") print(f"[BARON] Widget timeout: {exc}", flush=True) await page.wait_for_timeout(15_000) except Exception as exc: write_status("error", error=str(exc)) print(f"[BARON] Bridge error: {exc}", flush=True) await page.wait_for_timeout(15_000) async def main(): if not OUTPUT_FILE.exists(): write_json_atomic(OUTPUT_FILE, []) write_status("starting", strike_count=0) await run_bridge() if __name__ == "__main__": try: asyncio.run(main()) except KeyboardInterrupt: write_status("stopped") print("\n[BARON] Bridge stopped.", flush=True)