diff --git a/app/ais_stream.py b/app/ais_stream.py new file mode 100644 index 0000000..0c27cae --- /dev/null +++ b/app/ais_stream.py @@ -0,0 +1,102 @@ +"""AISStream WebSocket worker — server-side only. + +aisstream.io forbids browser clients. Connect from the FastAPI/ingest +process, upsert last-known positions, and expose them via GET /api/vessels. + +Idle (no crash) when AISSTREAM_API_KEY is unset. Reconnect with jittered +backoff and resend the full subscription within 3 seconds of each connect. +""" + +from __future__ import annotations + +import asyncio +import json +import logging +import os +import random + +from config import AISSTREAM_API_KEY, AISSTREAM_BBOX +from keystore import get_api_key +from live_layers import transform_ais_frame, upsert_vessel + +logger = logging.getLogger("osint.aisstream") + +WS_URL = "wss://stream.aisstream.io/v0/stream" +FILTER_TYPES = [ + "PositionReport", + "StandardClassBPositionReport", + "ExtendedClassBPositionReport", + "ShipStaticData", +] + + +def _parse_boxes(raw: str) -> list[list[list[float]]]: + """Env format: minlat,minlon,maxlat,maxlon[; ...]. AIS wants [[lat,lon],[lat,lon]].""" + boxes = [] + for chunk in (raw or "").split(";"): + parts = [p.strip() for p in chunk.split(",") if p.strip()] + if len(parts) != 4: + continue + minlat, minlon, maxlat, maxlon = (float(p) for p in parts) + boxes.append([[minlat, minlon], [maxlat, maxlon]]) + return boxes or [[[24.0, -125.0], [50.0, -66.0]]] + + +async def _resolve_key() -> str: + return (os.getenv("AISSTREAM_API_KEY") or AISSTREAM_API_KEY + or (await get_api_key("AISSTREAM_API_KEY")) or "").strip() + + +async def run_ais_worker() -> None: + """Long-lived reconnect loop. Safe to spawn as an asyncio task.""" + try: + import websockets + except ImportError: + logger.warning("websockets package not installed — AIS worker disabled") + return + + backoff = 2.0 + while True: + key = await _resolve_key() + if not key: + logger.warning( + "AISSTREAM_API_KEY not set — AIS ingest idle. " + "Create a free key at https://aisstream.io/account" + ) + await asyncio.sleep(60) + continue + boxes = _parse_boxes(AISSTREAM_BBOX) + try: + async with websockets.connect( + WS_URL, + max_size=2 ** 22, + ping_interval=20, + ping_timeout=20, + compression="deflate", + ) as ws: + sub = { + "APIKey": key, + "BoundingBoxes": boxes, + "FilterMessageTypes": FILTER_TYPES, + } + await ws.send(json.dumps(sub)) + logger.info("AISStream subscribed (%d bbox(es))", len(boxes)) + backoff = 2.0 + async for raw in ws: + if isinstance(raw, bytes): + raw = raw.decode("utf-8", errors="replace") + try: + frame = json.loads(raw) + except json.JSONDecodeError: + continue + marker = transform_ais_frame(frame) + if marker: + await upsert_vessel(marker) + except asyncio.CancelledError: + raise + except Exception: # noqa: BLE001 + logger.exception("AISStream disconnected") + delay = backoff + random.uniform(0, 1.5) + logger.info("AISStream reconnect in %.1fs", delay) + await asyncio.sleep(delay) + backoff = min(60.0, backoff * 1.7) diff --git a/app/camera_config.py b/app/camera_config.py index 7774a53..fd12ab6 100644 --- a/app/camera_config.py +++ b/app/camera_config.py @@ -12,6 +12,10 @@ import os # Formats: Insecam-style HTML, plain-text URL lists, ALERTWest JSON. # NOTE: docker-compose always defines CAMERA_SOURCE_URLS (empty when no .env), # so os.getenv()'s default would never apply — use `or` semantics instead. +CALTRANS_CCTV_URLS = tuple( + f"https://cwwp2.dot.ca.gov/data/d{n}/cctv/cctvStatusD{n:02d}.json" + for n in range(1, 13) +) _DEFAULT_SOURCE_URL = ",".join(( # Publicly published open-camera list (markdown bullets of stream URLs). "https://raw.githubusercontent.com/fury999io/public-ip-cams/main/README.md", @@ -19,6 +23,8 @@ _DEFAULT_SOURCE_URL = ",".join(( "https://api.cdn.prod.alertwest.com/api/getCameraDataByLoc", # Curated global outdoor streams (HLS / YouTube / JPEG) — VDOT, MDSHA, etc. "https://raw.githubusercontent.com/willytop8/Live-Environment-Streams/main/streams.geojson", + # Official Caltrans CWWP2 JPEG + HLS CCTV (districts 1–12). + *CALTRANS_CCTV_URLS, )) CAMERA_SOURCE_URLS = [ u.strip() diff --git a/app/camera_scraper.py b/app/camera_scraper.py index 444d226..ad5dc91 100644 --- a/app/camera_scraper.py +++ b/app/camera_scraper.py @@ -326,6 +326,59 @@ def parse_alertwest_json(text: str, source_name: str) -> list[dict]: return out +def parse_caltrans_json(text: str, source_name: str) -> list[dict]: + """Parse Caltrans CWWP2 cctvStatus JSON (districts 1–12, same schema). + + Skip cameras that are not in service. Store the JPEG still as snapshot_url + (map thumbs) and the HLS playlist as source_url when present. Never RTSP. + """ + try: + payload = json.loads(text) + except (json.JSONDecodeError, ValueError): + return [] + rows = payload.get("data") if isinstance(payload, dict) else None + if not isinstance(rows, list): + return [] + out: list[dict] = [] + for row in rows: + cctv = (row or {}).get("cctv") or row or {} + loc = cctv.get("location") or {} + in_service = str(cctv.get("inService") or "").strip().lower() + if in_service not in ("true", "1", "yes"): + continue + img = ((cctv.get("imageData") or {}).get("static") or {}) + jpeg = (img.get("currentImageURL") or img.get("currentImageUrl") or "").strip() + hls = str((cctv.get("imageData") or {}).get("streamingVideoURL") or "").strip() + if hls.lower().startswith("rtsp://") or jpeg.lower().startswith("rtsp://"): + continue + if not jpeg and not hls: + continue + try: + lat = float(loc["latitude"]) if loc.get("latitude") not in (None, "") else None + lon = float(loc["longitude"]) if loc.get("longitude") not in (None, "") else None + except (TypeError, ValueError): + lat = lon = None + bits = [ + loc.get("locationName"), + loc.get("route") and f"SR-{loc.get('route')}", + loc.get("nearbyPlace") or loc.get("county"), + ] + name = ", ".join(str(b) for b in bits if b) + ident = hls or jpeg + dtype = "hls" if hls else "http" + out.append({ + "source_url": ident, + "snapshot_url": jpeg or hls, + "discovery_source": "caltrans", + "location_lat": lat, + "location_lon": lon, + "location_name": name or None, + "vendor": "Caltrans", + "device_type": dtype, + }) + return out + + def parse_live_streams_geojson(text: str, source_name: str) -> list[dict]: """Parse willytop8/Live-Environment-Streams GeoJSON. @@ -435,7 +488,9 @@ async def scrape_source(client: RateLimitedClient, geo: Geocoder, name = urlparse(src_url).netloc ctype = resp.headers.get("content-type", "") body = resp.text - if ("getCameraDataByLoc" in src_url + if "cwwp2.dot.ca.gov" in src_url or "cctvStatus" in src_url: + cams = parse_caltrans_json(body, name) + elif ("getCameraDataByLoc" in src_url or ("json" in ctype and '"locs"' in body[:4000] and '"cams"' in body[:8000])): cams = parse_alertwest_json(body, name) elif (src_url.endswith(".geojson") or src_url.endswith("/streams.geojson") diff --git a/app/config.py b/app/config.py index 4d734ea..be6ffec 100644 --- a/app/config.py +++ b/app/config.py @@ -47,7 +47,8 @@ MINIO_SECURE = os.getenv("MINIO_SECURE", "false").lower() not in ("false", "0", # warning and stays idle (no crash). FIRMS_MAP_KEY = os.getenv("FIRMS_MAP_KEY", "") # NRT VIIRS S-NPP active fire/hotspot detection (375m). -FIRMS_DATASET = os.getenv("FIRMS_DATASET", "VIIRS_SNPP_NRT") +# NASA stops Suomi NPP product delivery on 2026-11-01 — default to NOAA-20. +FIRMS_DATASET = os.getenv("FIRMS_DATASET", "VIIRS_NOAA20_NRT") # Area bounding box as "minlon,minlat,maxlon,maxlat". Default covers most of # the inhabited globe; narrow it (e.g. CONUS "-125,24,-66,50") to shrink # payloads and the Postgres write volume. @@ -60,3 +61,23 @@ FIRMS_INTERVAL = int(os.getenv("FIRMS_INTERVAL", "900")) FIRMS_DAYS = int(os.getenv("FIRMS_DAYS", "2")) # Outbound HTTP timeout for the FIRMS CSV download. FIRMS_TIMEOUT = float(os.getenv("FIRMS_TIMEOUT", "60")) +# Comma-separated FIRMS products to dual-write. S-NPP delivery ends 2026-11-01; +# default to NOAA-20 + NOAA-21 NRT. FIRMS_DATASET is still honored when +# FIRMS_DATASETS is unset (empty string means "use FIRMS_DATASET only"). +_FIRMS_DATASETS_RAW = os.getenv("FIRMS_DATASETS", "VIIRS_NOAA20_NRT,VIIRS_NOAA21_NRT") +FIRMS_DATASETS = [d.strip() for d in _FIRMS_DATASETS_RAW.split(",") if d.strip()] or [FIRMS_DATASET] + +# Identifying User-Agent for NWS / Amtraker / Nominatim (mandatory on some APIs). +OSINT_USER_AGENT = os.getenv( + "OSINT_USER_AGENT", "osint-dashboard/1.0 (self-hosted; lancewalters94@gmail.com)" +) + +# AISStream (server-side WebSocket only). Idle when unset. +AISSTREAM_API_KEY = os.getenv("AISSTREAM_API_KEY", "") +# Bounding box(es) as minlat,minlon,maxlat,maxlon — note lat/lon order (AISStream). +# Default: CONUS coasts + Great Lakes, not the world. +AISSTREAM_BBOX = os.getenv("AISSTREAM_BBOX", "24,-125,50,-66") +# Run the AIS worker inside the dashboard process (default on so vessels work +# without the ingest profile). Set 0 if the ingester owns the only connection. +AISSTREAM_IN_APP = os.getenv("AISSTREAM_IN_APP", "1").lower() in ("1", "true", "yes") +AISSTREAM_IN_INGEST = os.getenv("AISSTREAM_IN_INGEST", "0").lower() in ("1", "true", "yes") diff --git a/app/fire_sources.py b/app/fire_sources.py index 0da0593..838d974 100644 --- a/app/fire_sources.py +++ b/app/fire_sources.py @@ -34,6 +34,7 @@ import nats from config import ( FIRMS_BBOX, FIRMS_DATASET, + FIRMS_DATASETS, FIRMS_DAYS, FIRMS_TIMEOUT, NATS_URL, @@ -180,26 +181,30 @@ async def ingest_fires(bbox: str | None = None) -> int: return 0 area = bbox or FIRMS_BBOX - url = FIRMS_AREA_CSV.format( - key=map_key, dataset=FIRMS_DATASET, bbox=area, days=FIRMS_DAYS - ) - + datasets = FIRMS_DATASETS or [FIRMS_DATASET] + total_published = 0 async with httpx.AsyncClient(timeout=FIRMS_TIMEOUT) as client: - resp = await client.get(url) - resp.raise_for_status() - text = resp.text - - # FIRMS returns HTTP 200 with a plain-text error for some failure modes - # (bad key, invalid bbox); surface the first line for debuggability. - if "latitude" not in text.lower()[:4096]: - first_line = text.strip().splitlines()[0][:200] if text.strip() else "(empty)" - logger.warning("FIRMS CSV download returned no hotspot data (%s)", first_line) - return 0 - - points = parse_firms_csv(text) - published = await publish_fire_batch(points) - logger.info( - "FIRMS: fetched %d hotspot(s) for bbox=%s (%s), published %d", - len(points), area, FIRMS_DATASET, published, - ) - return published + for dataset in datasets: + url = FIRMS_AREA_CSV.format( + key=map_key, dataset=dataset, bbox=area, days=FIRMS_DAYS + ) + resp = await client.get(url) + resp.raise_for_status() + text = resp.text + # FIRMS returns HTTP 200 with a plain-text error for some failure modes + # (bad key, invalid bbox); surface the first line for debuggability. + if "latitude" not in text.lower()[:4096]: + first_line = text.strip().splitlines()[0][:200] if text.strip() else "(empty)" + logger.warning( + "FIRMS CSV download returned no hotspot data for %s (%s)", + dataset, first_line, + ) + continue + points = parse_firms_csv(text) + published = await publish_fire_batch(points) + total_published += published + logger.info( + "FIRMS: fetched %d hotspot(s) for bbox=%s (%s), published %d", + len(points), area, dataset, published, + ) + return total_published diff --git a/app/keystore.py b/app/keystore.py index 1f84924..e2e5355 100644 --- a/app/keystore.py +++ b/app/keystore.py @@ -67,6 +67,19 @@ KEY_REGISTRY: dict[str, dict] = { "pattern": r"^\d{8,10}:[0-9A-Za-z_-]{35}$", "example": "123456789:AA… (bot token from @BotFather)", }, + "AISSTREAM_API_KEY": { + "description": "AISStream WebSocket key — live vessel positions (server-side only).", + "pattern": r"^.{8,}$", + "example": "key from https://aisstream.io/account (GitHub login)", + }, + "OPENSKY_CLIENT_ID": { + "description": "OpenSky OAuth client id — optional ADS-B fallback (unused until enabled).", + "example": "client id from opensky-network.org account", + }, + "OPENSKY_CLIENT_SECRET": { + "description": "OpenSky OAuth client secret — optional ADS-B fallback.", + "example": "client secret from the OpenSky account page", + }, } # Any stored key must at least be a sane UPPER_SNAKE name. diff --git a/app/live_layers.py b/app/live_layers.py new file mode 100644 index 0000000..3e214c6 --- /dev/null +++ b/app/live_layers.py @@ -0,0 +1,612 @@ +"""Live map overlays: parsers, TTL cache, and upstream fetchers. + +Moving objects (aircraft, vessels, trains) and alerts are vectors. Radar / +GIBS fire tiles are rasters served directly to the browser — this module +only returns tile *templates* and GeoJSON/JSON for vectors. + +Third-party APIs that leak keys, lack CORS, or rate-limit by IP are fetched +here (FastAPI), never from Leaflet. Viewport bbox only; never a global ADS-B +or AIS poll. +""" + +from __future__ import annotations + +import asyncio +import math +import time +from datetime import datetime, timezone +from typing import Any, Awaitable, Callable + +import httpx + +from config import OSINT_USER_AGENT + +MARKER_FIELDS = ("id", "lat", "lon", "heading", "speed", "label", "extra") + +ADSB_LOL_BASE = "https://api.adsb.lol" +AMTRAKER_TRAINS = "https://api.amtraker.com/v3/trains" +RAINVIEWER_MAPS = "https://api.rainviewer.com/public/weather-maps.json" +NWS_ALERTS = "https://api.weather.gov/alerts/active" +IEM_SBW = "https://mesonet.agron.iastate.edu/geojson/sbw.geojson" +WFIGS_INCIDENTS = ( + "https://services3.arcgis.com/T4QMspbfLg3qTGWY/arcgis/rest/services/" + "WFIGS_Incident_Locations_Current/FeatureServer/0/query" +) +WFIGS_PERIMETERS = ( + "https://services3.arcgis.com/T4QMspbfLg3qTGWY/arcgis/rest/services/" + "WFIGS_Interagency_Perimeters_Current/FeatureServer/0/query" +) +NHC_STORMS = "https://www.nhc.noaa.gov/CurrentStorms.json" + +IEM_NEXRAD = "https://mesonet.agron.iastate.edu/cache/tile.py/1.0.0/nexrad-n0q/{z}/{x}/{y}.png" +GIBS_THERMAL = ( + "https://gibs.earthdata.nasa.gov/wmts/epsg3857/best/" + "VIIRS_SNPP_Thermal_Anomalies_375m_All/default/{time}/" + "GoogleMapsCompatible_Level9/{z}/{y}/{x}.png" +) + +CONUS = (-125.0, 24.0, -66.0, 50.0) +MAX_RADIUS_NM = 150 +DEFAULT_LIMIT = 2000 + +_HTTP_TIMEOUT = 25.0 +_COMPASS = { + "N": 0, "NE": 45, "E": 90, "SE": 135, + "S": 180, "SW": 225, "W": 270, "NW": 315, + "NNE": 22, "ENE": 67, "ESE": 112, "SSE": 157, + "SSW": 202, "WSW": 247, "WNW": 292, "NNW": 337, +} + +_cache: dict[str, tuple[float, Any]] = {} +_cache_lock = asyncio.Lock() + +# Last-known AIS positions (MMSI -> marker). Filled by ais_stream worker. +vessel_last_known: dict[str, dict] = {} +vessel_lock = asyncio.Lock() + + +def overlay_catalog() -> dict: + """Tile templates + vector endpoint map for the layer panel. No secrets.""" + today = datetime.now(timezone.utc).date().isoformat() + return { + "radar_iem": { + "id": "radar_iem", + "title": "IEM NEXRAD (CONUS)", + "kind": "raster", + "tileUrl": IEM_NEXRAD, + "maxZoom": 18, + "attribution": "Iowa Environmental Mesonet", + }, + "radar_rainviewer": { + "id": "radar_rainviewer", + "title": "RainViewer (global)", + "kind": "raster", + "tileUrl": None, # filled from /api/map/radar frames + "maxZoom": 7, + "attribution": 'Weather data by RainViewer', + }, + "gibs_thermal": { + "id": "gibs_thermal", + "title": "GIBS VIIRS thermal anomalies", + "kind": "raster", + "tileUrl": GIBS_THERMAL.replace("{time}", today), + "timeTemplate": GIBS_THERMAL, + "maxZoom": 9, + "attribution": "NASA GIBS / EOSDIS", + }, + "nws_alerts": {"id": "nws_alerts", "kind": "geojson", "endpoint": "/api/weather-alerts"}, + "wfigs_incidents": {"id": "wfigs_incidents", "kind": "points", "endpoint": "/api/fire-incidents"}, + "wfigs_perimeters": {"id": "wfigs_perimeters", "kind": "geojson", "endpoint": "/api/fire-perimeters"}, + "aircraft": {"id": "aircraft", "kind": "points", "endpoint": "/api/aircraft"}, + "vessels": {"id": "vessels", "kind": "points", "endpoint": "/api/vessels"}, + "trains": {"id": "trains", "kind": "points", "endpoint": "/api/trains"}, + "storms": {"id": "storms", "kind": "points", "endpoint": "/api/storms"}, + } + + +def parse_bbox(bbox: str) -> tuple[float, float, float, float]: + """Parse 'minlon,minlat,maxlon,maxlat' into four floats.""" + parts = [p.strip() for p in (bbox or "").split(",")] + if len(parts) != 4: + raise ValueError("bbox must be 'minlon,minlat,maxlon,maxlat'") + try: + minlon, minlat, maxlon, maxlat = (float(p) for p in parts) + except ValueError as exc: + raise ValueError("bbox values must be floats") from exc + return minlon, minlat, maxlon, maxlat + + +def bbox_center_radius_nm( + minlon: float, minlat: float, maxlon: float, maxlat: float, +) -> tuple[float, float, int]: + """Viewport center + half-diagonal radius in nautical miles, clamped ≤ 150.""" + lat = (minlat + maxlat) / 2.0 + lon = (minlon + maxlon) / 2.0 + # Half the diagonal of the box, in nm (1 deg lat ≈ 60 nm). + dlat = abs(maxlat - minlat) / 2.0 + dlon = abs(maxlon - minlon) / 2.0 + km = _haversine_km(lat, lon, lat + dlat, lon + dlon) + nm = km / 1.852 + radius = max(1, min(MAX_RADIUS_NM, int(math.ceil(nm)))) + return lat, lon, radius + + +def _haversine_km(lat1: float, lon1: float, lat2: float, lon2: float) -> float: + r = 6371.0 + p1, p2 = math.radians(lat1), math.radians(lat2) + dphi = math.radians(lat2 - lat1) + dlmb = math.radians(lon2 - lon1) + a = math.sin(dphi / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(dlmb / 2) ** 2 + return 2 * r * math.asin(min(1.0, math.sqrt(a))) + + +def to_marker( + id_: str, + lat: float | None, + lon: float | None, + heading: float | None = None, + speed: float | None = None, + label: str | None = None, + extra: dict | None = None, +) -> dict: + return { + "id": str(id_), + "lat": lat, + "lon": lon, + "heading": heading, + "speed": speed, + "label": label or str(id_), + "extra": extra or {}, + } + + +def filter_points_bbox( + points: list[dict], + minlon: float, minlat: float, maxlon: float, maxlat: float, + limit: int = DEFAULT_LIMIT, +) -> list[dict]: + out = [] + for p in points: + lat, lon = p.get("lat"), p.get("lon") + if lat is None or lon is None: + continue + if minlat <= lat <= maxlat and minlon <= lon <= maxlon: + out.append(p) + if len(out) >= limit: + break + return out + + +def _f(value: object) -> float | None: + if value is None or value == "": + return None + try: + return float(value) + except (TypeError, ValueError): + return None + + +def _heading(value: object) -> float | None: + if value is None or value == "": + return None + if isinstance(value, str): + key = value.strip().upper() + if key in _COMPASS: + return float(_COMPASS[key]) + num = _f(value) + if num is None: + return None + if num < 0 or num > 360: + return None + return num + + +def transform_adsb_lol(payload: dict | list | None) -> list[dict]: + """Map ADSB.lol v2 aircraft list to shared markers. Dedup on hex.""" + if payload is None: + return [] + if isinstance(payload, list): + aircraft = payload + else: + aircraft = payload.get("ac") or payload.get("aircraft") or [] + seen: set[str] = set() + out: list[dict] = [] + for ac in aircraft: + hex_id = str(ac.get("hex") or "").strip().lower() + lat, lon = _f(ac.get("lat")), _f(ac.get("lon")) + if not hex_id or lat is None or lon is None: + continue + if hex_id in seen: + continue + seen.add(hex_id) + flight = str(ac.get("flight") or "").strip() or hex_id + out.append(to_marker( + hex_id, lat, lon, + heading=_heading(ac.get("track")), + speed=_f(ac.get("gs")), + label=flight, + extra={ + "hex": hex_id, + "reg": ac.get("r"), + "type": ac.get("t"), + "alt_baro": ac.get("alt_baro"), + "squawk": ac.get("squawk"), + "emergency": ac.get("emergency"), + "category": ac.get("category"), + "seen_pos": ac.get("seen_pos"), + "src": "adsb.lol", + }, + )) + return out + + +def transform_amtraker(payload: dict | None) -> list[dict]: + """Flatten Amtraker `{trainNum: [Train, ...]}` to one marker per trainID.""" + if not payload or not isinstance(payload, dict): + return [] + out: list[dict] = [] + for _num, trains in payload.items(): + if not isinstance(trains, list): + continue + for tr in trains: + if not isinstance(tr, dict): + continue + lat, lon = _f(tr.get("lat")), _f(tr.get("lon")) + if lat is None or lon is None: + continue + tid = str(tr.get("trainID") or tr.get("trainId") or "").strip() + tnum = str(tr.get("trainNum") or _num) + route = str(tr.get("routeName") or "").strip() + label = f"{route} #{tnum}".strip() if route else f"Train {tnum}" + late = tr.get("late") + if late is None: + late = tr.get("lateMin") + out.append(to_marker( + tid or tnum, lat, lon, + heading=_heading(tr.get("heading")), + speed=_f(tr.get("velocity") or tr.get("speed")), + label=label, + extra={ + "route": route, + "trainNum": tnum, + "late_min": late, + "iconColor": tr.get("iconColor"), + "stations": tr.get("stations") or [], + "src": "amtraker", + }, + )) + return out + + +def transform_ais_frame(frame: dict | None) -> dict | None: + """Decode one AISStream JSON envelope to a marker (or None if unusable).""" + if not frame or not isinstance(frame, dict): + return None + meta = frame.get("MetaData") or {} + mmsi = meta.get("MMSI") or meta.get("mmsi") + if mmsi is None: + return None + name = str(meta.get("ShipName") or meta.get("shipName") or "").strip() + lat = _f(meta.get("Latitude") if "Latitude" in meta else meta.get("latitude")) + lon = _f(meta.get("Longitude") if "Longitude" in meta else meta.get("longitude")) + msg = frame.get("Message") or {} + pos = ( + msg.get("PositionReport") + or msg.get("StandardClassBPositionReport") + or msg.get("ExtendedClassBPositionReport") + or {} + ) + extra: dict[str, Any] = {"src": "aisstream", "mmsi": mmsi} + if frame.get("MessageType") == "ShipStaticData": + static = msg.get("ShipStaticData") or {} + dest = str(static.get("Destination") or static.get("destination") or "").strip() + extra["dest"] = dest + extra["static"] = True + if not name: + name = str(static.get("Name") or static.get("name") or "").strip() + if lat is None or lon is None: + # Static-only update: caller merges onto last-known by MMSI. + return to_marker(str(mmsi), None, None, label=name or str(mmsi), extra=extra) + heading = pos.get("TrueHeading") + if heading in (511, 511.0, None): + heading = pos.get("Cog") + sog = pos.get("Sog") + navstat = pos.get("NavigationalStatus") + extra["navstat"] = navstat + extra["cog"] = pos.get("Cog") + extra["dest"] = extra.get("dest") + if lat is None or lon is None: + return None + return to_marker( + str(mmsi), lat, lon, + heading=_heading(heading), + speed=_f(sog), + label=name or str(mmsi), + extra=extra, + ) + + +def transform_wfigs_incidents(fc: dict | None) -> list[dict]: + if not fc: + return [] + out: list[dict] = [] + for feat in fc.get("features") or []: + geom = feat.get("geometry") or {} + coords = geom.get("coordinates") or [] + if len(coords) < 2: + continue + lon, lat = _f(coords[0]), _f(coords[1]) + if lat is None or lon is None: + continue + props = feat.get("properties") or {} + name = str(props.get("IncidentName") or "Incident").strip() + out.append(to_marker( + name, lat, lon, + label=name, + extra={ + "acres": props.get("IncidentSize"), + "contained": props.get("PercentContained"), + "state": props.get("POOState"), + "category": props.get("IncidentTypeCategory"), + "cause": props.get("FireCause"), + "discovered": props.get("FireDiscoveryDateTime"), + "src": "wfigs", + }, + )) + return out + + +def transform_nhc_storms(payload: dict | None) -> list[dict]: + if not payload: + return [] + storms = payload.get("activeStorms") or [] + class_label = { + "TD": "Tropical Depression", + "TS": "Tropical Storm", + "HU": "Hurricane", + "STD": "Subtropical Depression", + "STS": "Subtropical Storm", + "PTC": "Potential Tropical Cyclone", + "PC": "Post-Tropical Cyclone", + } + out: list[dict] = [] + for s in storms: + lat = _f(s.get("latitudeNumeric") or s.get("lat")) + lon = _f(s.get("longitudeNumeric") or s.get("lon")) + if lat is None or lon is None: + continue + sid = str(s.get("id") or s.get("binNumber") or "").strip() + name = str(s.get("name") or "Storm").strip() + klass = str(s.get("classification") or "").upper() + kind = class_label.get(klass, klass or "Storm") + out.append(to_marker( + sid or name, lat, lon, + heading=_heading(s.get("movementDir")), + speed=_f(s.get("movementSpeed")), + label=f"{kind} {name}".strip(), + extra={ + "classification": klass, + "intensity_kt": s.get("intensity"), + "src": "nhc", + }, + )) + return out + + +def rainviewer_tile_url(host: str, path: str, size: int = 256, color: int = 2, + smooth: int = 1, snow: int = 1) -> str: + host = host.rstrip("/") + path = path if path.startswith("/") else f"/{path}" + return f"{host}{path}/{size}/{{z}}/{{x}}/{{y}}/{color}/{smooth}_{snow}.png" + + +def _headers() -> dict[str, str]: + return {"User-Agent": OSINT_USER_AGENT, "Accept": "application/json"} + + +async def _ttl_get(key: str, ttl: float, factory: Callable[[], Awaitable[Any]]) -> Any: + now = time.monotonic() + hit = _cache.get(key) + if hit and now - hit[0] < ttl: + return hit[1] + async with _cache_lock: + hit = _cache.get(key) + if hit and time.monotonic() - hit[0] < ttl: + return hit[1] + value = await factory() + _cache[key] = (time.monotonic(), value) + return value + + +async def _get_json(url: str, params: dict | None = None) -> Any: + async with httpx.AsyncClient(timeout=_HTTP_TIMEOUT, follow_redirects=True, + headers=_headers()) as client: + resp = await client.get(url, params=params) + resp.raise_for_status() + return resp.json() + + +async def fetch_aircraft(bbox: str, limit: int = DEFAULT_LIMIT) -> list[dict]: + minlon, minlat, maxlon, maxlat = parse_bbox(bbox) + lat, lon, radius = bbox_center_radius_nm(minlon, minlat, maxlon, maxlat) + cache_key = f"adsb:{lat:.2f}:{lon:.2f}:{radius}" + + async def _load(): + url = f"{ADSB_LOL_BASE}/v2/lat/{lat:.4f}/lon/{lon:.4f}/dist/{radius}" + return transform_adsb_lol(await _get_json(url)) + + rows = await _ttl_get(cache_key, 8.0, _load) + return filter_points_bbox(rows, minlon, minlat, maxlon, maxlat, limit) + + +async def fetch_trains(bbox: str | None, limit: int = DEFAULT_LIMIT) -> list[dict]: + async def _load(): + return transform_amtraker(await _get_json(AMTRAKER_TRAINS)) + + rows = await _ttl_get("amtraker:trains", 20.0, _load) + if bbox: + minlon, minlat, maxlon, maxlat = parse_bbox(bbox) + return filter_points_bbox(rows, minlon, minlat, maxlon, maxlat, limit) + return rows[:limit] + + +async def fetch_vessels(bbox: str | None, limit: int = DEFAULT_LIMIT) -> list[dict]: + async with vessel_lock: + rows = [dict(v) for v in vessel_last_known.values() + if v.get("lat") is not None and v.get("lon") is not None] + if bbox: + minlon, minlat, maxlon, maxlat = parse_bbox(bbox) + return filter_points_bbox(rows, minlon, minlat, maxlon, maxlat, limit) + return rows[:limit] + + +async def upsert_vessel(marker: dict) -> None: + """Merge an AIS marker into last-known by MMSI. Static-only updates names.""" + vid = str(marker.get("id") or "") + if not vid: + return + async with vessel_lock: + prev = vessel_last_known.get(vid, {}) + extra = {**(prev.get("extra") or {}), **(marker.get("extra") or {})} + lat = marker.get("lat") if marker.get("lat") is not None else prev.get("lat") + lon = marker.get("lon") if marker.get("lon") is not None else prev.get("lon") + label = marker.get("label") + if not label or label == vid: + label = prev.get("label") or vid + vessel_last_known[vid] = { + **to_marker( + vid, lat, lon, + heading=marker.get("heading") if marker.get("heading") is not None else prev.get("heading"), + speed=marker.get("speed") if marker.get("speed") is not None else prev.get("speed"), + label=label, + extra=extra, + ), + "seen_at": datetime.now(timezone.utc).isoformat(), + } + + +def _wfigs_params(bbox: str | None) -> dict: + params = { + "where": "1=1", + "outSR": "4326", + "f": "geojson", + "resultRecordCount": 2000, + } + if bbox: + minlon, minlat, maxlon, maxlat = parse_bbox(bbox) + params["geometry"] = f"{minlon},{minlat},{maxlon},{maxlat}" + params["geometryType"] = "esriGeometryEnvelope" + params["inSR"] = "4326" + params["spatialRel"] = "esriSpatialRelIntersects" + return params + + +async def fetch_fire_incidents(bbox: str | None, limit: int = DEFAULT_LIMIT) -> list[dict]: + params = _wfigs_params(bbox) + params["outFields"] = ( + "IncidentName,IncidentSize,FireDiscoveryDateTime,POOState," + "PercentContained,IncidentTypeCategory,FireCause" + ) + + async def _load(): + return transform_wfigs_incidents(await _get_json(WFIGS_INCIDENTS, params)) + + rows = await _ttl_get(f"wfigs:inc:{bbox or 'all'}", 600.0, _load) + return rows[:limit] + + +async def fetch_fire_perimeters(bbox: str | None) -> dict: + params = _wfigs_params(bbox) + params["outFields"] = ( + "poly_IncidentName,poly_GISAcres,attr_IncidentSize," + "attr_PercentContained,attr_FireDiscoveryDateTime" + ) + + async def _load(): + return await _get_json(WFIGS_PERIMETERS, params) + + fc = await _ttl_get(f"wfigs:per:{bbox or 'all'}", 600.0, _load) + if not isinstance(fc, dict): + return {"type": "FeatureCollection", "features": []} + return fc + + +async def fetch_weather_alerts(area: str | None, bbox: str | None) -> dict: + """Cached NWS active alerts + IEM storm-based warning polygons.""" + + async def _load(): + nws_params: dict[str, str] = {"status": "actual"} + if area: + nws_params["area"] = area.upper() + elif bbox: + minlon, minlat, maxlon, maxlat = parse_bbox(bbox) + nws_params["bbox"] = f"{minlon},{minlat},{maxlon},{maxlat}" + nws_fc: dict = {"features": []} + sbw_fc: dict = {"features": []} + try: + nws_fc = await _get_json(NWS_ALERTS, nws_params) + except Exception: + nws_fc = {"features": []} + try: + sbw_fc = await _get_json(IEM_SBW) + except Exception: + sbw_fc = {"features": []} + features = [] + for feat in nws_fc.get("features") or []: + props = feat.get("properties") or {} + props["source"] = "nws" + feat["properties"] = props + features.append(feat) + for feat in sbw_fc.get("features") or []: + props = feat.get("properties") or {} + props["source"] = "iem-sbw" + # IEM uses `ps` (phenomenon) / `wfo`; map a display event. + if "event" not in props: + props["event"] = props.get("ps") or "Storm-based warning" + feat["properties"] = props + if bbox: + # Cheap reject: skip if no geometry; keep otherwise (polygons). + if not feat.get("geometry"): + continue + features.append(feat) + return {"type": "FeatureCollection", "features": features} + + key = f"alerts:{area or ''}:{bbox or ''}" + return await _ttl_get(key, 30.0, _load) + + +async def fetch_radar_meta() -> dict: + async def _load(): + data = await _get_json(RAINVIEWER_MAPS) + host = data.get("host") or "https://tilecache.rainviewer.com" + radar = (data.get("radar") or {}) + past = radar.get("past") or [] + nowcast = radar.get("nowcast") or [] + frames = [] + for fr in past + nowcast: + path = fr.get("path") + if not path: + continue + frames.append({ + "time": fr.get("time"), + "path": path, + "tileUrl": rainviewer_tile_url(host, path), + }) + latest = frames[-1]["tileUrl"] if frames else None + return { + "provider": "rainviewer", + "host": host, + "tileUrl": latest, + "frames": frames, + "attribution": "Weather data by RainViewer", + "iemTileUrl": IEM_NEXRAD, + } + + return await _ttl_get("radar:rv", 300.0, _load) + + +async def fetch_storms() -> list[dict]: + async def _load(): + return transform_nhc_storms(await _get_json(NHC_STORMS)) + + return await _ttl_get("nhc:storms", 300.0, _load) diff --git a/app/main.py b/app/main.py index 2969af0..3b4defe 100644 --- a/app/main.py +++ b/app/main.py @@ -44,6 +44,11 @@ from ingestor import ingest_event, fetch_and_process from sources import ingest_rss_feed, ingest_gdelt, ingest_earthquakes, ingest_social_signals from fire_sources import ingest_fires from keystore import KeyFormatError, delete_key, list_keys, set_key +from live_layers import ( + fetch_aircraft, fetch_fire_incidents, fetch_fire_perimeters, + fetch_radar_meta, fetch_storms, fetch_trains, fetch_vessels, + fetch_weather_alerts, overlay_catalog, parse_bbox, +) logging.basicConfig(level=logging.INFO) logger = structlog.get_logger("osint.dashboard") @@ -136,6 +141,10 @@ async def startup(): head) before uvicorn starts, so they don't run nested inside the event loop. """ await init_extensions() + from config import AISSTREAM_IN_APP + if AISSTREAM_IN_APP: + from ais_stream import run_ais_worker + asyncio.create_task(run_ais_worker()) # ── Feed Sources ────────────────────────────────────────────────────────── @@ -1013,6 +1022,13 @@ async def index(): # ── NASA GIBS basemap map tab ───────────────────────────────────────────── +def _parse_bbox_query(bbox: str) -> tuple[float, float, float, float]: + try: + return parse_bbox(bbox) + except ValueError as exc: + raise HTTPException(422, str(exc)) from exc + + @app.get("/api/map/layers") async def map_layers(): """Curated NASA GIBS raster basemap layers for the map tab. @@ -1024,9 +1040,122 @@ async def map_layers(): format — tile image extension (jpeg|png) has_time — whether the layer has a Time dimension (=> date selector) max_zoom — highest native zoom served by that tile matrix set + + ``overlays`` lists toggleable live feeds (radar tiles, aircraft, …). """ from gibs_map import MAP_LAYERS - return {"layers": MAP_LAYERS} + return {"layers": MAP_LAYERS, "overlays": overlay_catalog()} + + +def _upstream_or_502(exc: Exception, name: str): + logger.warning("live_layer_upstream_failed", layer=name, error=str(exc)) + raise HTTPException(502, f"{name} upstream unavailable: {exc}") from exc + + +@app.get("/api/map/radar") +async def map_radar(): + """RainViewer frame list + IEM NEXRAD tile template. Browser fetches tiles.""" + try: + return await fetch_radar_meta() + except Exception as exc: + _upstream_or_502(exc, "radar") + + +@app.get("/api/aircraft") +async def list_aircraft( + bbox: str = Query(..., description="minlon,minlat,maxlon,maxlat"), + limit: int = Query(2000, ge=1, le=5000), +): + """Viewport ADS-B last-known (ADSB.lol). Requires bbox; radius clamped ≤ 150 nm.""" + _parse_bbox_query(bbox) + try: + return await fetch_aircraft(bbox, limit) + except ValueError as exc: + raise HTTPException(422, str(exc)) from exc + except Exception as exc: + _upstream_or_502(exc, "aircraft") + + +@app.get("/api/trains") +async def list_trains( + bbox: str | None = Query(None, description="minlon,minlat,maxlon,maxlat"), + limit: int = Query(2000, ge=1, le=5000), +): + """Amtrak / Brightline / VIA last-known (Amtraker). Bbox optional (~200 rows).""" + if bbox: + _parse_bbox_query(bbox) + try: + return await fetch_trains(bbox, limit) + except ValueError as exc: + raise HTTPException(422, str(exc)) from exc + except Exception as exc: + _upstream_or_502(exc, "trains") + + +@app.get("/api/vessels") +async def list_vessels( + bbox: str | None = Query(None, description="minlon,minlat,maxlon,maxlat"), + limit: int = Query(2000, ge=1, le=5000), +): + """AIS last-known from the server-side AISStream worker. Empty without a key.""" + if bbox: + _parse_bbox_query(bbox) + try: + return await fetch_vessels(bbox, limit) + except ValueError as exc: + raise HTTPException(422, str(exc)) from exc + + +@app.get("/api/fire-incidents") +async def list_fire_incidents( + bbox: str | None = Query(None), + limit: int = Query(2000, ge=1, le=5000), +): + """NIFC WFIGS current incident points.""" + if bbox: + _parse_bbox_query(bbox) + try: + return await fetch_fire_incidents(bbox, limit) + except Exception as exc: + _upstream_or_502(exc, "fire-incidents") + + +@app.get("/api/fire-perimeters") +async def list_fire_perimeters(bbox: str | None = Query(None)): + """NIFC WFIGS current wildfire perimeters (GeoJSON).""" + if bbox: + _parse_bbox_query(bbox) + try: + return await fetch_fire_perimeters(bbox) + except Exception as exc: + _upstream_or_502(exc, "fire-perimeters") + + +@app.get("/api/weather-alerts") +async def list_weather_alerts( + area: str | None = Query(None, description="US state two-letter code, e.g. NC"), + bbox: str | None = Query(None, description="minlon,minlat,maxlon,maxlat"), +): + """Cached NWS active alerts + IEM storm-based warning polygons. + + Named ``/api/weather-alerts`` so it does not collide with dashboard + ``/api/alerts`` (entity/keyword alert records). + """ + if bbox: + _parse_bbox_query(bbox) + try: + return await fetch_weather_alerts(area, bbox) + except Exception as exc: + _upstream_or_502(exc, "weather-alerts") + + +@app.get("/api/storms") +async def list_storms(): + """NHC active tropical cyclones.""" + try: + return await fetch_storms() + except Exception as exc: + _upstream_or_502(exc, "storms") @app.get("/api/map/times") diff --git a/app/requirements.txt b/app/requirements.txt index e5ee103..508bcb6 100644 --- a/app/requirements.txt +++ b/app/requirements.txt @@ -10,3 +10,4 @@ httpx>=0.28 feedparser>=6.0 python-dateutil>=2.9 structlog>=24.4 +websockets>=14 diff --git a/app/run_ingester.py b/app/run_ingester.py index 1dfdd66..56e0062 100644 --- a/app/run_ingester.py +++ b/app/run_ingester.py @@ -24,7 +24,7 @@ import sys sys.path.insert(0, sys_path) -from config import NATS_URL, FIRMS_INTERVAL, FIRMS_DATASET # noqa: E402 +from config import NATS_URL, FIRMS_INTERVAL, FIRMS_DATASET, AISSTREAM_IN_INGEST # noqa: E402 from sources import ingest_rss_feed, ingest_gdelt, ingest_earthquakes # noqa: E402 from fire_sources import ingest_fires # noqa: E402 from ingestor import ingest_event, start_nats_consumer # noqa: E402 @@ -112,6 +112,9 @@ async def main() -> None: # Fire ingest only starts once FIRMS_MAP_KEY is set (ingest_fires logs # and idles otherwise). tasks.append(asyncio.create_task(fire_loop())) + if AISSTREAM_IN_INGEST: + from ais_stream import run_ais_worker # noqa: E402 + tasks.append(asyncio.create_task(run_ais_worker())) await asyncio.gather(producer_loop(), consumer_loop(), *tasks) diff --git a/app/static/index.html b/app/static/index.html index 9afe27b..e06d8d3 100644 --- a/app/static/index.html +++ b/app/static/index.html @@ -256,7 +256,7 @@ /* ── Layer control panel (dark HUD) ── */ #layer-panel { position: absolute; top: 58px; left: 12px; z-index: 600; - width: 252px; max-height: calc(100% - 74px); overflow: hidden; + width: 268px; max-height: calc(100% - 74px); overflow: hidden; background: rgba(6,11,20,0.86); backdrop-filter: blur(8px); border: 1px solid var(--line-hi); border-radius: 6px; box-shadow: 0 0 22px rgba(53,224,255,0.14), 0 0 2px rgba(53,224,255,0.4); @@ -289,6 +289,11 @@ .lp-dot.weather { background: var(--amber); box-shadow: 0 0 7px var(--amber); } .lp-dot.flights { background: #7dd3fc; box-shadow: 0 0 7px #7dd3fc; } .lp-dot.vessels { background: #2dd4bf; box-shadow: 0 0 7px #2dd4bf; } + .lp-dot.radar { background: #38bdf8; box-shadow: 0 0 7px #38bdf8; } + .lp-dot.thermal { background: #f97316; box-shadow: 0 0 7px #f97316; } + .lp-dot.wfigs { background: #ef4444; box-shadow: 0 0 7px #ef4444; } + .lp-dot.trains { background: #c084fc; box-shadow: 0 0 7px #c084fc; } + .lp-dot.storms { background: #f472b6; box-shadow: 0 0 7px #f472b6; } .lp-count { font-family: 'Share Tech Mono', monospace; font-size: 0.7rem; color: var(--cyan); } .lp-sub { display: flex; justify-content: space-between; align-items: center; gap: 0.5rem; } .lp-opacity { display: flex; align-items: center; gap: 0.4rem; font-size: 0.62rem; color: var(--muted); text-transform: uppercase; letter-spacing: 0.05em; } @@ -684,19 +689,62 @@
Geolocated events from the ingest pipeline. Color = source type.
-
+
- + + IEM +
+
Opacity 70%
+
IEM NEXRAD over CONUS, RainViewer elsewhere. Tiles load in the browser.
+
+
+
+ + 0
-
+
- + + VIIRS
-
+
- + + 0 +
+
+
+
+ + 0 +
+
+
+
+ + 0 +
+
+
+
+ + 0 +
+
Amtraker · ODC-By
+
+
+
+ + 0 +
+
Needs AISSTREAM_API_KEY in Keys.
+
+
+
+ + 0
@@ -1564,6 +1612,19 @@ let mapLayers = []; let mapDomain = null; let mapLatest = null; let mapAttributionAdded = null; +let extraAttribs = new Set(); +let radarLayer = null, radarOn = true, radarOpacity = 0.7, radarMeta = null, radarTimer = null; +let thermalLayer = null, thermalOn = false; +let wxAlertsGroup = null, wxAlertsOn = true; +let perimGroup = null, perimOn = true; +let incidentsGroup = null, incidentsOn = false; +let acGroup = null, acOn = true; +let trainsGroup = null, trainsOn = true; +let vesselsGroup = null, vesselsOn = false; +let stormsGroup = null, stormsOn = true; +let overlayReq = {ac:0, trains:0, vessels:0, alerts:0, perim:0, incidents:0, storms:0}; +let moveDebounce = null; +const pointCanvas = () => L.canvas({ padding: 0.5 }); async function initMap() { if (mapInitStarted) { if (map) setTimeout(() => map.invalidateSize(), 60); return; } @@ -1637,9 +1698,13 @@ async function initMap() { map.on('dragstart', () => { if (camPopupOpen) map.closePopup(); }); map.on('moveend', () => { if (camPopupOpen) return; // only the popup's own autopan now - if (firesOn) loadFires(); - if (camsOn) loadCams(); - if (blipsOn) loadBlips(); + if (moveDebounce) clearTimeout(moveDebounce); + moveDebounce = setTimeout(() => { + if (firesOn) loadFires(); + if (camsOn) loadCams(); + if (blipsOn) loadBlips(); + refreshLiveOverlays(); + }, 300); }); await mapLayerChanged(); // Overlays default ON (checkboxes in the layer panel + saved settings) @@ -1649,9 +1714,19 @@ async function initMap() { blipsOn = document.getElementById('lp-blips-on').checked; firesSince = document.getElementById('lp-fires-since').value; blipsSince = document.getElementById('lp-blips-since').value; + radarOn = document.getElementById('lp-radar-on').checked; + wxAlertsOn = document.getElementById('lp-alerts-on').checked; + thermalOn = document.getElementById('lp-thermal-on').checked; + perimOn = document.getElementById('lp-perim-on').checked; + incidentsOn = document.getElementById('lp-incidents-on').checked; + acOn = document.getElementById('lp-ac-on').checked; + trainsOn = document.getElementById('lp-trains-on').checked; + vesselsOn = document.getElementById('lp-vessels-on').checked; + stormsOn = document.getElementById('lp-storms-on').checked; if (firesOn) loadFires(); if (camsOn) loadCams(); if (blipsOn) loadBlips(); + refreshLiveOverlays(); } catch(e) { hint.textContent = `Failed to load map layers: ${e.message || e}`; console.error('Map init failed', e); @@ -1757,6 +1832,7 @@ function mapDateChanged() { document.getElementById('map-hint').textContent += ' ⚠ outside published windows (GIBS serves nearest-time)'; } } + if (thermalOn) loadThermal(); } function mapGoLatest() { document.getElementById('map-date').value = mapLatest || ''; mapDateChanged(); } function mapGoDays(n) { @@ -2064,6 +2140,309 @@ async function loadBlips() { } } +/* ── Live overlays (radar / alerts / WFIGS / aircraft / trains / AIS / NHC) ── */ +function refreshLiveOverlays() { + if (!map) return; + if (radarOn) loadRadar(); + if (thermalOn) loadThermal(); + if (wxAlertsOn) loadWxAlerts(); + if (perimOn) loadPerimeters(); + if (incidentsOn) loadIncidents(); + if (acOn) loadAircraft(); + if (trainsOn) loadTrains(); + if (vesselsOn) loadVessels(); + if (stormsOn) loadStorms(); +} +function addExtraAttrib(html) { + if (!map || !html || extraAttribs.has(html)) return; + extraAttribs.add(html); + map.attributionControl.addAttribution(html); +} +function dropLayer(ref) { + if (ref && map && map.hasLayer(ref)) map.removeLayer(ref); + return null; +} +function intersectsConus() { + if (!map) return false; + const b = map.getBounds(); + const west = Math.max(b.getWest(), -125); + const east = Math.min(b.getEast(), -66); + const south = Math.max(b.getSouth(), 24); + const north = Math.min(b.getNorth(), 50); + return west < east && south < north && map.getZoom() >= 4; +} +function severityColor(sev) { + const s = String(sev || '').toLowerCase(); + if (s === 'extreme' || s === 'warning') return '#ff5d5d'; + if (s === 'severe') return '#fb923c'; + if (s === 'moderate') return '#ffb454'; + if (s === 'minor') return '#53f0a5'; + return '#35e0ff'; +} +function altColor(alt) { + const a = Number(alt) || 0; + if (a >= 35000) return '#e0f2fe'; + if (a >= 20000) return '#7dd3fc'; + if (a >= 10000) return '#38bdf8'; + if (a >= 1000) return '#fbbf24'; + return '#fb923c'; +} +function pointPopup(p) { + const extra = p.extra || {}; + const rows = Object.keys(extra).filter(k => k !== 'stations' && extra[k] != null && extra[k] !== '') + .slice(0, 8) + .map(k => `${esc(k)}${esc(extra[k])}`).join(''); + return `
${esc(p.label || p.id)}${rows}
`; +} +function renderPoints(existing, points, colorFn, cluster) { + if (existing) map.removeLayer(existing); + const zoom = map.getZoom(); + const useCluster = cluster && (zoom < 7 || points.length > 200); + const group = useCluster + ? L.markerClusterGroup({ maxClusterRadius: 48, showCoverageOnHover: false, spiderfyOnMaxZoom: true, chunkedLoading: true }) + : L.layerGroup(); + const renderer = pointCanvas(); + points.forEach(p => { + if (p.lat == null || p.lon == null) return; + const col = colorFn(p); + const heading = Number(p.heading); + const m = L.circleMarker([p.lat, p.lon], { + radius: 5, color: col, fillColor: col, fillOpacity: 0.9, weight: 1, + renderer, + }).bindPopup(pointPopup(p)); + if (!Number.isNaN(heading)) m.setStyle({ className: 'hdg' }); + group.addLayer(m); + }); + group.addTo(map); + return group; +} +async function toggleRadar() { + radarOn = document.getElementById('lp-radar-on').checked; + if (radarOn) await loadRadar(); + else { radarLayer = dropLayer(radarLayer); if (radarTimer) { clearInterval(radarTimer); radarTimer = null; } } +} +function setRadarOpacity(v) { + radarOpacity = v / 100; + document.getElementById('lp-radar-val').textContent = `${Math.round(v)}%`; + if (radarLayer) radarLayer.setOpacity(radarOpacity); +} +async function loadRadar() { + if (!map) return; + try { + if (!radarMeta) { + const r = await fetch(`${API}/api/map/radar`); + radarMeta = await r.json(); + addExtraAttrib('Weather data by RainViewer'); + addExtraAttrib('Iowa Environmental Mesonet'); + if (!radarTimer) radarTimer = setInterval(() => { radarMeta = null; if (radarOn) loadRadar(); }, 5 * 60 * 1000); + } + const useIem = intersectsConus(); + const url = useIem + ? (radarMeta.iemTileUrl || 'https://mesonet.agron.iastate.edu/cache/tile.py/1.0.0/nexrad-n0q/{z}/{x}/{y}.png') + : (radarMeta.tileUrl); + if (!url) return; + document.getElementById('lp-radar-count').textContent = useIem ? 'IEM' : 'RV'; + if (radarLayer) { + radarLayer.setUrl(url); + radarLayer.setOpacity(radarOpacity); + } else { + radarLayer = L.tileLayer(url, { opacity: radarOpacity, maxZoom: 12, maxNativeZoom: useIem ? 18 : 7, attribution: '' }).addTo(map); + } + } catch (e) { + console.error('Radar load failed', e); + document.getElementById('lp-radar-count').textContent = 'err'; + } +} +async function toggleThermal() { + thermalOn = document.getElementById('lp-thermal-on').checked; + if (thermalOn) loadThermal(); + else thermalLayer = dropLayer(thermalLayer); +} +function loadThermal() { + if (!map) return; + const time = (document.getElementById('map-date') && document.getElementById('map-date').value) + || new Date().toISOString().slice(0, 10); + const url = `https://gibs.earthdata.nasa.gov/wmts/epsg3857/best/VIIRS_SNPP_Thermal_Anomalies_375m_All/default/${time}/GoogleMapsCompatible_Level9/{z}/{y}/{x}.png`; + addExtraAttrib('NASA GIBS thermal anomalies'); + if (thermalLayer) thermalLayer.setUrl(url); + else thermalLayer = L.tileLayer(url, { opacity: 0.85, maxNativeZoom: 9, maxZoom: 12, attribution: '' }).addTo(map); +} +async function toggleWxAlerts() { + wxAlertsOn = document.getElementById('lp-alerts-on').checked; + if (wxAlertsOn) await loadWxAlerts(); + else wxAlertsGroup = dropLayer(wxAlertsGroup); +} +async function loadWxAlerts() { + if (!map) return; + const req = ++overlayReq.alerts; + try { + const r = await fetch(`${API}/api/weather-alerts?bbox=${currentBBox()}`); + const fc = await r.json(); + if (req !== overlayReq.alerts) return; + wxAlertsGroup = dropLayer(wxAlertsGroup); + const feats = fc.features || []; + wxAlertsGroup = L.geoJSON(fc, { + style: (f) => ({ + color: severityColor((f.properties || {}).severity), + weight: 2, fillOpacity: 0.18, + }), + onEachFeature: (f, layer) => { + const p = f.properties || {}; + layer.bindPopup(`
${esc(p.source || 'alert')}
${esc(p.event || p.headline || 'Alert')}
${esc(p.severity || '')} · ${esc(p.areaDesc || p.wfo || '')}
`); + }, + }).addTo(map); + document.getElementById('lp-alerts-count').textContent = feats.length.toLocaleString(); + addExtraAttrib('NWS / IEM storm-based warnings'); + } catch (e) { + console.error('Alerts load failed', e); + document.getElementById('lp-alerts-count').textContent = 'err'; + } +} +async function togglePerimeters() { + perimOn = document.getElementById('lp-perim-on').checked; + if (perimOn) await loadPerimeters(); + else perimGroup = dropLayer(perimGroup); +} +async function loadPerimeters() { + if (!map) return; + const req = ++overlayReq.perim; + try { + const r = await fetch(`${API}/api/fire-perimeters?bbox=${currentBBox()}`); + const fc = await r.json(); + if (req !== overlayReq.perim) return; + perimGroup = dropLayer(perimGroup); + const feats = fc.features || []; + perimGroup = L.geoJSON(fc, { + style: (f) => { + const acres = Number((f.properties || {}).poly_GISAcres || (f.properties || {}).attr_IncidentSize || 0); + return { color: acres > 10000 ? '#ef4444' : '#fb923c', weight: 2, fillOpacity: 0.25, fillColor: '#fb923c' }; + }, + onEachFeature: (f, layer) => { + const p = f.properties || {}; + const name = p.poly_IncidentName || p.IncidentName || 'Fire'; + const acres = p.poly_GISAcres || p.attr_IncidentSize || '?'; + layer.bindPopup(`
${esc(name)}
${esc(acres)} acres · ${esc(p.attr_PercentContained || '')}% contained
`); + }, + }).addTo(map); + document.getElementById('lp-perim-count').textContent = feats.length.toLocaleString(); + addExtraAttrib('NIFC WFIGS'); + } catch (e) { + console.error('Perimeters load failed', e); + document.getElementById('lp-perim-count').textContent = 'err'; + } +} +async function toggleIncidents() { + incidentsOn = document.getElementById('lp-incidents-on').checked; + if (incidentsOn) await loadIncidents(); + else incidentsGroup = dropLayer(incidentsGroup); +} +async function loadIncidents() { + if (!map) return; + const req = ++overlayReq.incidents; + try { + const r = await fetch(`${API}/api/fire-incidents?bbox=${currentBBox()}`); + const pts = await r.json(); + if (req !== overlayReq.incidents) return; + incidentsGroup = renderPoints(incidentsGroup, pts, () => '#ef4444', false); + document.getElementById('lp-incidents-count').textContent = pts.length.toLocaleString(); + addExtraAttrib('NIFC WFIGS'); + } catch (e) { + console.error('Incidents load failed', e); + document.getElementById('lp-incidents-count').textContent = 'err'; + } +} +async function toggleAircraft() { + acOn = document.getElementById('lp-ac-on').checked; + if (acOn) await loadAircraft(); + else acGroup = dropLayer(acGroup); +} +async function loadAircraft() { + if (!map) return; + if (map.getZoom() <= 3) { + document.getElementById('lp-ac-count').textContent = 'zoom'; + return; + } + const req = ++overlayReq.ac; + try { + const r = await fetch(`${API}/api/aircraft?bbox=${currentBBox()}`); + const pts = await r.json(); + if (req !== overlayReq.ac) return; + acGroup = renderPoints(acGroup, Array.isArray(pts) ? pts : [], p => altColor((p.extra || {}).alt_baro), true); + document.getElementById('lp-ac-count').textContent = (pts.length || 0).toLocaleString(); + addExtraAttrib('ADSB.lol ODbL'); + } catch (e) { + console.error('Aircraft load failed', e); + document.getElementById('lp-ac-count').textContent = 'err'; + } +} +async function toggleTrains() { + trainsOn = document.getElementById('lp-trains-on').checked; + if (trainsOn) await loadTrains(); + else trainsGroup = dropLayer(trainsGroup); +} +async function loadTrains() { + if (!map) return; + const req = ++overlayReq.trains; + try { + const r = await fetch(`${API}/api/trains?bbox=${currentBBox()}`); + const pts = await r.json(); + if (req !== overlayReq.trains) return; + trainsGroup = renderPoints(trainsGroup, Array.isArray(pts) ? pts : [], p => (p.extra || {}).iconColor || '#c084fc', false); + document.getElementById('lp-trains-count').textContent = (pts.length || 0).toLocaleString(); + addExtraAttrib('Amtraker'); + } catch (e) { + console.error('Trains load failed', e); + document.getElementById('lp-trains-count').textContent = 'err'; + } +} +async function toggleVessels() { + vesselsOn = document.getElementById('lp-vessels-on').checked; + if (vesselsOn) await loadVessels(); + else vesselsGroup = dropLayer(vesselsGroup); +} +async function loadVessels() { + if (!map) return; + if (map.getZoom() <= 3) { + document.getElementById('lp-vessels-count').textContent = 'zoom'; + return; + } + const req = ++overlayReq.vessels; + try { + const r = await fetch(`${API}/api/vessels?bbox=${currentBBox()}`); + const pts = await r.json(); + if (req !== overlayReq.vessels) return; + vesselsGroup = renderPoints(vesselsGroup, Array.isArray(pts) ? pts : [], p => { + const sog = Number((p.speed) || 0); + return sog > 0.5 ? '#2dd4bf' : '#64748b'; + }, true); + document.getElementById('lp-vessels-count').textContent = (pts.length || 0).toLocaleString(); + addExtraAttrib('AISStream'); + } catch (e) { + console.error('Vessels load failed', e); + document.getElementById('lp-vessels-count').textContent = 'err'; + } +} +async function toggleStorms() { + stormsOn = document.getElementById('lp-storms-on').checked; + if (stormsOn) await loadStorms(); + else stormsGroup = dropLayer(stormsGroup); +} +async function loadStorms() { + if (!map) return; + const req = ++overlayReq.storms; + try { + const r = await fetch(`${API}/api/storms`); + const pts = await r.json(); + if (req !== overlayReq.storms) return; + stormsGroup = renderPoints(stormsGroup, Array.isArray(pts) ? pts : [], () => '#f472b6', false); + document.getElementById('lp-storms-count').textContent = (pts.length || 0).toLocaleString(); + addExtraAttrib('NHC'); + } catch (e) { + console.error('Storms load failed', e); + document.getElementById('lp-storms-count').textContent = 'err'; + } +} + /* ═══════════════ INITIAL LOAD ═══════════════ */ initNav(); initSettings(); diff --git a/docker-compose.yml b/docker-compose.yml index bb12741..e03a016 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -80,9 +80,14 @@ services: # ── NASA FIRMS active fires ── INGEST_FIRES: ${INGEST_FIRES:-1} FIRMS_MAP_KEY: ${FIRMS_MAP_KEY:-} - FIRMS_DATASET: ${FIRMS_DATASET:-VIIRS_SNPP_NRT} + FIRMS_DATASET: ${FIRMS_DATASET:-VIIRS_NOAA20_NRT} + FIRMS_DATASETS: ${FIRMS_DATASETS:-VIIRS_NOAA20_NRT,VIIRS_NOAA21_NRT} FIRMS_BBOX: ${FIRMS_BBOX:--180,-60,180,75} FIRMS_INTERVAL: ${FIRMS_INTERVAL:-900} + OSINT_USER_AGENT: ${OSINT_USER_AGENT:-osint-dashboard/1.0 (self-hosted)} + AISSTREAM_API_KEY: ${AISSTREAM_API_KEY:-} + AISSTREAM_BBOX: ${AISSTREAM_BBOX:-24,-125,50,-66} + AISSTREAM_IN_INGEST: ${AISSTREAM_IN_INGEST:-0} command: ["python", "app/run_ingester.py"] entrypoint: ["python", "app/run_ingester.py"] @@ -110,8 +115,13 @@ services: MINIO_SECURE: ${MINIO_SECURE:-false} # ── NASA FIRMS (for the /api/ingest/fires trigger endpoint) ── FIRMS_MAP_KEY: ${FIRMS_MAP_KEY:-} - FIRMS_DATASET: ${FIRMS_DATASET:-VIIRS_SNPP_NRT} + FIRMS_DATASET: ${FIRMS_DATASET:-VIIRS_NOAA20_NRT} + FIRMS_DATASETS: ${FIRMS_DATASETS:-VIIRS_NOAA20_NRT,VIIRS_NOAA21_NRT} FIRMS_BBOX: ${FIRMS_BBOX:--180,-60,180,75} + OSINT_USER_AGENT: ${OSINT_USER_AGENT:-osint-dashboard/1.0 (self-hosted)} + AISSTREAM_API_KEY: ${AISSTREAM_API_KEY:-} + AISSTREAM_BBOX: ${AISSTREAM_BBOX:-24,-125,50,-66} + AISSTREAM_IN_APP: ${AISSTREAM_IN_APP:-1} ports: - "127.0.0.1:8000:8000" healthcheck: diff --git a/tests/test_api_live_layers.py b/tests/test_api_live_layers.py new file mode 100644 index 0000000..a9e2c1d --- /dev/null +++ b/tests/test_api_live_layers.py @@ -0,0 +1,38 @@ +"""API contract tests for live overlay endpoints (no DB required).""" + +import asyncio + +import httpx + +from live_layers import overlay_catalog +from main import app + +BASE = "http://test" + + +async def _get(path: str) -> httpx.Response: + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url=BASE) as client: + return await client.get(path) + + +def test_map_layers_includes_overlays(): + body = asyncio.run(_get("/api/map/layers")).json() + assert "layers" in body + overlays = body["overlays"] + for key in ("radar_iem", "radar_rainviewer", "gibs_thermal", + "aircraft", "vessels", "trains", "nws_alerts", + "wfigs_incidents", "wfigs_perimeters"): + assert key in overlays + assert overlay_catalog()["radar_iem"]["tileUrl"].startswith("https://mesonet") + + +def test_aircraft_requires_bbox(): + resp = asyncio.run(_get("/api/aircraft")) + assert resp.status_code == 422 + + +def test_vessels_empty_without_ais_key(): + resp = asyncio.run(_get("/api/vessels")) + assert resp.status_code == 200 + assert resp.json() == [] diff --git a/tests/test_fire_sources.py b/tests/test_fire_sources.py index ec13b1e..3a3e4af 100644 --- a/tests/test_fire_sources.py +++ b/tests/test_fire_sources.py @@ -109,6 +109,7 @@ def test_ingest_fires_uses_keystore_key(monkeypatch): return False async def get(self, url): + captured.setdefault("urls", []).append(url) captured["url"] = url return FakeResp() @@ -122,8 +123,10 @@ def test_ingest_fires_uses_keystore_key(monkeypatch): monkeypatch.setattr("fire_sources.publish_fire_batch", fake_publish) - assert asyncio.run(ingest_fires()) == 5 + assert asyncio.run(ingest_fires()) == 10 # NOAA-20 + NOAA-21 dual-write assert "a" * 32 in captured["url"] + assert any("VIIRS_NOAA20_NRT" in u for u in captured["urls"]) + assert any("VIIRS_NOAA21_NRT" in u for u in captured["urls"]) def _async_return(value): diff --git a/tests/test_live_layers.py b/tests/test_live_layers.py new file mode 100644 index 0000000..2214a5c --- /dev/null +++ b/tests/test_live_layers.py @@ -0,0 +1,246 @@ +"""Unit tests for live map-layer mappers (aircraft, trains, AIS, WFIGS, Caltrans).""" + +from live_layers import ( + MARKER_FIELDS, + bbox_center_radius_nm, + filter_points_bbox, + parse_bbox, + rainviewer_tile_url, + to_marker, + transform_adsb_lol, + transform_ais_frame, + transform_amtraker, + transform_nhc_storms, + transform_wfigs_incidents, +) + +from camera_scraper import parse_caltrans_json + + +def test_parse_bbox_and_radius_clamps_to_150_nm(): + minlon, minlat, maxlon, maxlat = parse_bbox("-84.5,33.8,-75.4,36.6") + assert (minlon, minlat, maxlon, maxlat) == (-84.5, 33.8, -75.4, 36.6) + lat, lon, radius = bbox_center_radius_nm(minlon, minlat, maxlon, maxlat) + assert 35.0 < lat < 35.4 + assert -80.1 < lon < -79.8 + assert 1 <= radius <= 150 + + +def test_parse_bbox_rejects_malformed(): + try: + parse_bbox("1,2,3") + assert False, "expected ValueError" + except ValueError: + pass + + +def test_transform_adsb_lol_maps_shared_marker_contract(): + payload = { + "ac": [ + { + "hex": "a1b2c3", + "flight": "AAL123 ", + "r": "N123AA", + "t": "B738", + "lat": 35.88, + "lon": -78.79, + "alt_baro": 32000, + "gs": 430.2, + "track": 87.5, + "squawk": "1200", + "emergency": "none", + "category": "A3", + "seen_pos": 0.4, + }, + {"hex": "dead00", "flight": "NOFIX"}, # no coords → drop + ] + } + rows = transform_adsb_lol(payload) + assert len(rows) == 1 + m = rows[0] + assert set(MARKER_FIELDS).issubset(m) + assert m["id"] == "a1b2c3" + assert m["lat"] == 35.88 + assert m["lon"] == -78.79 + assert m["heading"] == 87.5 + assert m["speed"] == 430.2 + assert m["label"] == "AAL123" + assert m["extra"]["squawk"] == "1200" + assert m["extra"]["alt_baro"] == 32000 + + +def test_transform_amtraker_flattens_train_numbers(): + payload = { + "1": [ + { + "trainID": "1-9", + "trainNum": "1", + "routeName": "Sunset Limited", + "lat": 29.76, + "lon": -95.36, + "heading": 90, + "velocity": 45.0, + "late": 12, + "iconColor": "#ee3a43", + "stations": [{"name": "Houston", "status": "enroute"}], + } + ], + "5": [ + { + "trainID": "5-12", + "trainNum": "5", + "routeName": "California Zephyr", + "lat": 40.0, + "lon": -105.0, + "heading": "W", + "lateMin": 5, + "iconColor": "#005eb8", + } + ], + } + rows = transform_amtraker(payload) + assert {r["id"] for r in rows} == {"1-9", "5-12"} + sunset = next(r for r in rows if r["id"] == "1-9") + assert sunset["lat"] == 29.76 + assert sunset["label"] == "Sunset Limited #1" + assert sunset["speed"] == 45.0 + assert sunset["extra"]["late_min"] == 12 + assert sunset["extra"]["iconColor"] == "#ee3a43" + + +def test_transform_ais_position_report(): + frame = { + "MessageType": "PositionReport", + "MetaData": { + "MMSI": 366912810, + "ShipName": "EVER GIVEN", + "latitude": 36.9, + "longitude": -76.3, + }, + "Message": { + "PositionReport": { + "Sog": 12.4, + "Cog": 88.0, + "TrueHeading": 90, + "NavigationalStatus": 0, + } + }, + } + row = transform_ais_frame(frame) + assert row is not None + assert row["id"] == "366912810" + assert row["lat"] == 36.9 + assert row["lon"] == -76.3 + assert row["label"] == "EVER GIVEN" + assert row["speed"] == 12.4 + assert row["heading"] == 90 + assert row["extra"]["navstat"] == 0 + + +def test_transform_ais_ignores_non_position(): + assert transform_ais_frame({"MessageType": "Unknown"}) is None + + +def test_transform_wfigs_incidents_geojson(): + fc = { + "type": "FeatureCollection", + "features": [ + { + "type": "Feature", + "geometry": {"type": "Point", "coordinates": [-81.3, 28.5]}, + "properties": { + "IncidentName": "Foster Bridge", + "IncidentSize": 675, + "PercentContained": 100, + "POOState": "US-FL", + "IncidentTypeCategory": "WF", + "FireCause": "Human", + "FireDiscoveryDateTime": 1750000000000, + }, + } + ], + } + rows = transform_wfigs_incidents(fc) + assert len(rows) == 1 + assert rows[0]["label"] == "Foster Bridge" + assert rows[0]["lat"] == 28.5 + assert rows[0]["lon"] == -81.3 + assert rows[0]["extra"]["acres"] == 675 + assert rows[0]["extra"]["contained"] == 100 + + +def test_filter_points_bbox(): + pts = [ + to_marker("a", 35.0, -78.0, label="in"), + to_marker("b", 10.0, 20.0, label="out"), + ] + kept = filter_points_bbox(pts, -80, 33, -75, 37) + assert [p["id"] for p in kept] == ["a"] + + +def test_rainviewer_tile_url(): + url = rainviewer_tile_url( + host="https://tilecache.rainviewer.com", + path="/v2/radar/cb581daa2c0f", + ) + assert url == ( + "https://tilecache.rainviewer.com/v2/radar/cb581daa2c0f/256/{z}/{x}/{y}/2/1_1.png" + ) + + +def test_transform_nhc_storms(): + payload = { + "activeStorms": [ + { + "id": "al042026", + "name": "Dolly", + "classification": "TS", + "latitudeNumeric": 13.6, + "longitudeNumeric": -38.7, + "movementDir": 280, + "movementSpeed": 12, + "intensity": 35, + } + ] + } + rows = transform_nhc_storms(payload) + assert len(rows) == 1 + assert rows[0]["id"] == "al042026" + assert rows[0]["label"] == "Tropical Storm Dolly" + assert rows[0]["lat"] == 13.6 + + +def test_parse_caltrans_skips_oos_and_maps_jpeg_hls(): + payload = """ + {"data":[ + {"cctv":{ + "location":{ + "latitude":"37.8","longitude":"-122.4", + "locationName":"I-80 WB","nearbyPlace":"SF", + "district":"4","route":"80","county":"SF","direction":"W" + }, + "inService":"true", + "imageData":{ + "static":{"currentImageURL":"https://cwwp2.dot.ca.gov/data/d4/cctv/image/cam.jpg"}, + "streamingVideoURL":"https://wzmedia.dot.ca.gov/D4/cam.stream/playlist.m3u8" + } + }}, + {"cctv":{ + "location":{"latitude":"1","longitude":"2","locationName":"down"}, + "inService":"false", + "imageData":{"static":{"currentImageURL":"https://example.com/x.jpg"}} + }} + ]} + """ + cams = parse_caltrans_json(payload, "caltrans") + assert len(cams) == 1 + cam = cams[0] + assert cam["discovery_source"] == "caltrans" + assert cam["snapshot_url"].endswith("cam.jpg") + assert cam["source_url"].endswith("playlist.m3u8") + assert cam["device_type"] == "hls" + assert cam["location_lat"] == 37.8 + assert cam["location_lon"] == -122.4 + assert "I-80" in cam["location_name"] + assert "rtsp://" not in cam["source_url"].lower() + assert "rtsp://" not in cam["snapshot_url"].lower()