diff --git a/alembic/versions/005_phase2.py b/alembic/versions/005_phase2.py
new file mode 100644
index 0000000..277e30a
--- /dev/null
+++ b/alembic/versions/005_phase2.py
@@ -0,0 +1,196 @@
+"""phase 2: geofences, 1-min track CAGGs, fire/aircraft hits
+
+Revision ID: 005_phase2
+Revises: 004_camera_enum
+Create Date: 2026-08-28
+"""
+
+from alembic import op
+
+revision = "005_phase2"
+down_revision = "004_camera_enum"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+ op.execute("CREATE EXTENSION IF NOT EXISTS postgis")
+ op.execute("CREATE EXTENSION IF NOT EXISTS timescaledb")
+
+ op.execute("""
+ CREATE TABLE IF NOT EXISTS geofences (
+ id UUID PRIMARY KEY,
+ name TEXT NOT NULL,
+ geojson JSONB NOT NULL,
+ geom geometry(Polygon, 4326),
+ active INTEGER NOT NULL DEFAULT 1,
+ created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+ updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+ )
+ """)
+ op.execute("""
+ CREATE INDEX IF NOT EXISTS ix_geofences_geom
+ ON geofences USING gist (geom)
+ """)
+
+ op.execute("""
+ CREATE TABLE IF NOT EXISTS geofence_alerts (
+ id UUID PRIMARY KEY,
+ geofence_id UUID NOT NULL,
+ source_kind TEXT NOT NULL,
+ entity_id TEXT NOT NULL,
+ lat DOUBLE PRECISION,
+ lon DOUBLE PRECISION,
+ payload JSONB,
+ created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+ )
+ """)
+ op.execute("""
+ CREATE INDEX IF NOT EXISTS ix_geofence_alerts_created
+ ON geofence_alerts (created_at DESC)
+ """)
+
+ op.execute("""
+ CREATE TABLE IF NOT EXISTS vessel_positions (
+ mmsi TEXT NOT NULL,
+ ts TIMESTAMPTZ NOT NULL,
+ lat DOUBLE PRECISION NOT NULL,
+ lon DOUBLE PRECISION NOT NULL,
+ heading DOUBLE PRECISION,
+ speed DOUBLE PRECISION,
+ label TEXT,
+ extra JSONB,
+ PRIMARY KEY (mmsi, ts)
+ )
+ """)
+ op.execute("""
+ SELECT create_hypertable(
+ 'vessel_positions', 'ts', if_not_exists => TRUE
+ )
+ """)
+ op.execute("""
+ CREATE INDEX IF NOT EXISTS ix_vessel_positions_bbox
+ ON vessel_positions (lon, lat)
+ """)
+
+ op.execute("""
+ CREATE TABLE IF NOT EXISTS aircraft_positions (
+ hex TEXT NOT NULL,
+ ts TIMESTAMPTZ NOT NULL,
+ lat DOUBLE PRECISION NOT NULL,
+ lon DOUBLE PRECISION NOT NULL,
+ heading DOUBLE PRECISION,
+ speed DOUBLE PRECISION,
+ label TEXT,
+ extra JSONB,
+ PRIMARY KEY (hex, ts)
+ )
+ """)
+ op.execute("""
+ SELECT create_hypertable(
+ 'aircraft_positions', 'ts', if_not_exists => TRUE
+ )
+ """)
+ op.execute("""
+ CREATE INDEX IF NOT EXISTS ix_aircraft_positions_bbox
+ ON aircraft_positions (lon, lat)
+ """)
+
+ op.execute("""
+ CREATE TABLE IF NOT EXISTS fire_aircraft_hits (
+ id UUID NOT NULL,
+ fire_id TEXT NOT NULL,
+ fire_lat DOUBLE PRECISION NOT NULL,
+ fire_lon DOUBLE PRECISION NOT NULL,
+ aircraft_hex TEXT NOT NULL,
+ aircraft_type TEXT,
+ aircraft_lat DOUBLE PRECISION NOT NULL,
+ aircraft_lon DOUBLE PRECISION NOT NULL,
+ distance_mi DOUBLE PRECISION NOT NULL,
+ seen_at TIMESTAMPTZ NOT NULL,
+ PRIMARY KEY (fire_id, aircraft_hex, seen_at)
+ )
+ """)
+ op.execute("""
+ CREATE INDEX IF NOT EXISTS ix_fire_aircraft_hits_seen
+ ON fire_aircraft_hits (seen_at DESC)
+ """)
+
+ # 1-minute continuous aggregates (Timescale). last() keeps the newest
+ # sample in each bucket — the DVR slider reads these, not the raw table.
+ op.execute("""
+ DO $$
+ BEGIN
+ IF NOT EXISTS (
+ SELECT 1 FROM timescaledb_information.continuous_aggregates
+ WHERE view_name = 'vessel_tracks_1min'
+ ) THEN
+ EXECUTE $v$
+ CREATE MATERIALIZED VIEW vessel_tracks_1min
+ WITH (timescaledb.continuous) AS
+ SELECT time_bucket('1 minute', ts) AS bucket,
+ mmsi,
+ last(lat, ts) AS lat,
+ last(lon, ts) AS lon,
+ last(heading, ts) AS heading,
+ last(speed, ts) AS speed,
+ last(label, ts) AS label
+ FROM vessel_positions
+ GROUP BY bucket, mmsi
+ WITH NO DATA
+ $v$;
+ END IF;
+ IF NOT EXISTS (
+ SELECT 1 FROM timescaledb_information.continuous_aggregates
+ WHERE view_name = 'aircraft_tracks_1min'
+ ) THEN
+ EXECUTE $a$
+ CREATE MATERIALIZED VIEW aircraft_tracks_1min
+ WITH (timescaledb.continuous) AS
+ SELECT time_bucket('1 minute', ts) AS bucket,
+ hex,
+ last(lat, ts) AS lat,
+ last(lon, ts) AS lon,
+ last(heading, ts) AS heading,
+ last(speed, ts) AS speed,
+ last(label, ts) AS label
+ FROM aircraft_positions
+ GROUP BY bucket, hex
+ WITH NO DATA
+ $a$;
+ END IF;
+ END
+ $$;
+ """)
+ op.execute("""
+ DO $$
+ BEGIN
+ PERFORM add_continuous_aggregate_policy(
+ 'vessel_tracks_1min',
+ start_offset => INTERVAL '3 hours',
+ end_offset => INTERVAL '1 minute',
+ schedule_interval => INTERVAL '1 minute',
+ if_not_exists => TRUE
+ );
+ PERFORM add_continuous_aggregate_policy(
+ 'aircraft_tracks_1min',
+ start_offset => INTERVAL '3 hours',
+ end_offset => INTERVAL '1 minute',
+ schedule_interval => INTERVAL '1 minute',
+ if_not_exists => TRUE
+ );
+ EXCEPTION WHEN OTHERS THEN
+ NULL;
+ END
+ $$;
+ """)
+
+
+def downgrade() -> None:
+ op.execute("DROP MATERIALIZED VIEW IF EXISTS aircraft_tracks_1min CASCADE")
+ op.execute("DROP MATERIALIZED VIEW IF EXISTS vessel_tracks_1min CASCADE")
+ op.execute("DROP TABLE IF EXISTS fire_aircraft_hits")
+ op.execute("DROP TABLE IF EXISTS aircraft_positions")
+ op.execute("DROP TABLE IF EXISTS vessel_positions")
+ op.execute("DROP TABLE IF EXISTS geofence_alerts")
+ op.execute("DROP TABLE IF EXISTS geofences")
diff --git a/app/bg_jobs.py b/app/bg_jobs.py
new file mode 100644
index 0000000..8d14747
--- /dev/null
+++ b/app/bg_jobs.py
@@ -0,0 +1,103 @@
+"""Background masscan / ffmpeg — never block a FastAPI request on a scan.
+
+masscan is capped at 200 pps (home uplink saturates at 1k+). ffmpeg frame
+grabs are scheduled with asyncio.create_task and shared per URL.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import shutil
+from cachetools import TTLCache
+
+logger = logging.getLogger("osint.bg_jobs")
+
+MASSCAN_PPS_CAP = 200
+
+_masscan_task: asyncio.Task | None = None
+_ffmpeg_cache: TTLCache = TTLCache(maxsize=100, ttl=300)
+_ffmpeg_tasks: dict[str, asyncio.Task] = {}
+_FFMPEG = shutil.which("ffmpeg")
+
+
+def schedule_masscan_pass() -> bool:
+ """Kick one capped masscan pass. Returns False if a pass is already running."""
+ global _masscan_task
+ if _masscan_task is not None and not _masscan_task.done():
+ return False
+ _masscan_task = asyncio.create_task(_run_masscan_capped())
+ return True
+
+
+async def _run_masscan_capped() -> None:
+ import masscan_config as cfg
+ from run_masscan_service import _verify_excludefile, run_pass
+
+ orig = cfg.MASSCAN_RATE
+ if orig > MASSCAN_PPS_CAP:
+ logger.warning("capping masscan rate %s pps -> %s", orig, MASSCAN_PPS_CAP)
+ cfg.MASSCAN_RATE = MASSCAN_PPS_CAP
+ try:
+ _verify_excludefile()
+ await run_pass()
+ finally:
+ cfg.MASSCAN_RATE = orig
+
+
+def cached_ffmpeg_jpeg(url: str) -> bytes | None:
+ return _ffmpeg_cache.get(url)
+
+
+def schedule_ffmpeg_snapshot(url: str, timeout: float = 8.0) -> asyncio.Task:
+ """Start (or reuse) an ffmpeg JPEG grab. Caller may await the task."""
+ existing = _ffmpeg_tasks.get(url)
+ if existing is not None and not existing.done():
+ return existing
+ task = asyncio.create_task(_ffmpeg_grab_and_cache(url, timeout))
+ _ffmpeg_tasks[url] = task
+ return task
+
+
+async def _ffmpeg_grab(url: str, timeout: float = 8.0) -> bytes | None:
+ """Grab one JPEG frame. Isolated so tests can stub it."""
+ if not _FFMPEG or not url.lower().startswith("rtsp://"):
+ return None
+ cmd = [
+ _FFMPEG, "-hide_banner", "-loglevel", "error", "-nostdin",
+ "-rtsp_transport", "tcp",
+ "-timeout", "4000000",
+ "-i", url,
+ "-frames:v", "1",
+ "-f", "image2pipe", "-vcodec", "mjpeg",
+ "pipe:1",
+ ]
+ try:
+ proc = await asyncio.create_subprocess_exec(
+ *cmd,
+ stdout=asyncio.subprocess.PIPE,
+ stderr=asyncio.subprocess.DEVNULL,
+ )
+ except FileNotFoundError:
+ return None
+ try:
+ stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=timeout)
+ except asyncio.TimeoutError:
+ proc.kill()
+ try:
+ await proc.wait()
+ except Exception: # noqa: BLE001
+ pass
+ return None
+ if proc.returncode not in (0, None) or not stdout or len(stdout) < 64:
+ return None
+ if stdout[:2] != b"\xff\xd8":
+ return None
+ return stdout
+
+
+async def _ffmpeg_grab_and_cache(url: str, timeout: float) -> bytes | None:
+ data = await _ffmpeg_grab(url, timeout)
+ if data:
+ _ffmpeg_cache[url] = data
+ return data
diff --git a/app/camera_preview.py b/app/camera_preview.py
index b82dba0..2c89e77 100644
--- a/app/camera_preview.py
+++ b/app/camera_preview.py
@@ -137,38 +137,17 @@ async def probe_public_feed(host: str) -> str | None:
async def ffmpeg_snapshot(url: str, timeout: float = 8.0) -> bytes | None:
- """Grab a single JPEG frame from an RTSP URL. None if ffmpeg missing/fails."""
- if not _FFMPEG or not url.lower().startswith("rtsp://"):
- return None
- cmd = [
- _FFMPEG, "-hide_banner", "-loglevel", "error", "-nostdin",
- "-rtsp_transport", "tcp",
- "-timeout", "4000000", # 4s socket timeout, microseconds
- "-i", url,
- "-frames:v", "1",
- "-f", "image2pipe", "-vcodec", "mjpeg",
- "pipe:1",
- ]
- try:
- proc = await asyncio.create_subprocess_exec(
- *cmd,
- stdout=asyncio.subprocess.PIPE,
- stderr=asyncio.subprocess.DEVNULL,
- )
- except FileNotFoundError:
- return None
- try:
- stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=timeout)
- except asyncio.TimeoutError:
- proc.kill()
- try:
- await proc.wait()
- except Exception: # noqa: BLE001
- pass
- return None
- if proc.returncode not in (0, None) or not _looks_like_jpeg(stdout or b""):
- return None
- return stdout
+ """Grab a single JPEG frame from an RTSP URL. None if ffmpeg missing/fails.
+
+ The subprocess is scheduled via asyncio.create_task (shared per URL) so
+ concurrent popup clicks do not stack ffmpeg processes on the request path.
+ """
+ from bg_jobs import cached_ffmpeg_jpeg, schedule_ffmpeg_snapshot
+
+ hit = cached_ffmpeg_jpeg(url)
+ if hit:
+ return hit
+ return await schedule_ffmpeg_snapshot(url, timeout)
async def ffmpeg_mjpeg_stream(url: str):
diff --git a/app/fire_aircraft.py b/app/fire_aircraft.py
new file mode 100644
index 0000000..a2bf8de
--- /dev/null
+++ b/app/fire_aircraft.py
@@ -0,0 +1,186 @@
+"""Flag firefighting ADS-B aircraft within 20 miles of an active fire."""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+from uuid import uuid4
+
+from sqlalchemy import text
+
+from database import async_session
+from live_layers import _haversine_km
+
+
+RADIUS_MILES = 20.0
+RADIUS_KM = RADIUS_MILES * 1.609344
+_recent: dict[tuple[str, str], datetime] = {}
+_COOLDOWN = timedelta(minutes=5)
+
+# ICAO type designators commonly used on wildfire air tankers, scoopers,
+# helitack, and air-attack platforms. Uppercase; compared case-insensitively.
+FIREFIGHTER_ICAO = frozenset({
+ "AT802", "AT8T", "AT8P", "AT8B",
+ "C130", "C30J", "C130J",
+ "DC10", "MD10", "MD11", "MD87",
+ "B737", "B38M",
+ "CL2T", "CL215", "CL415", "CL5T",
+ "S64", "SK64",
+ "UH1", "UH1Y", "UH60", "H60", "S70",
+ "B412", "B212", "B205",
+ "AS50", "AS350", "A119", "A109",
+ "B350", "BE20",
+ "OV10",
+ "RJ85", "RJ1H", "B461", "B462", "B463",
+ "PC12",
+ "TBM7", "TBM8", "TBM9",
+ "C208",
+ "DH8D", "Q400",
+})
+
+
+def _icao(marker: dict) -> str:
+ extra = marker.get("extra") or {}
+ return str(extra.get("type") or extra.get("t") or "").strip().upper()
+
+
+def is_firefighter(marker: dict) -> bool:
+ return _icao(marker) in FIREFIGHTER_ICAO
+
+
+def _haversine_mi(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
+ return _haversine_km(lat1, lon1, lat2, lon2) / 1.609344
+
+
+def correlate_aircraft_to_fires(
+ fires: list[dict],
+ aircraft: list[dict],
+ radius_mi: float = RADIUS_MILES,
+) -> list[dict]:
+ hits: list[dict] = []
+ for fire in fires:
+ flat, flon = fire.get("lat"), fire.get("lon")
+ if flat is None or flon is None:
+ continue
+ fid = str(fire.get("id") or fire.get("label") or "fire")
+ for ac in aircraft:
+ if not is_firefighter(ac):
+ continue
+ alat, alon = ac.get("lat"), ac.get("lon")
+ if alat is None or alon is None:
+ continue
+ dist = _haversine_mi(float(flat), float(flon), float(alat), float(alon))
+ if dist > radius_mi:
+ continue
+ hits.append({
+ "fire_id": fid,
+ "fire_lat": float(flat),
+ "fire_lon": float(flon),
+ "aircraft_hex": str(ac.get("id")),
+ "aircraft_type": _icao(ac),
+ "aircraft_lat": float(alat),
+ "aircraft_lon": float(alon),
+ "distance_mi": round(dist, 2),
+ "label": ac.get("label") or ac.get("id"),
+ })
+ return hits
+
+
+async def persist_hits(hits: list[dict], seen_at: datetime | None = None) -> int:
+ if not hits:
+ return 0
+ ts = seen_at or datetime.now(timezone.utc)
+ n = 0
+ async with async_session() as session:
+ for h in hits:
+ try:
+ await session.execute(
+ text(
+ """
+ INSERT INTO fire_aircraft_hits
+ (id, fire_id, fire_lat, fire_lon,
+ aircraft_hex, aircraft_type, aircraft_lat, aircraft_lon,
+ distance_mi, seen_at)
+ VALUES (
+ CAST(:id AS uuid), :fire_id, :fire_lat, :fire_lon,
+ :aircraft_hex, :aircraft_type, :aircraft_lat, :aircraft_lon,
+ :distance_mi, :seen_at
+ )
+ ON CONFLICT (fire_id, aircraft_hex, seen_at) DO NOTHING
+ """
+ ),
+ {
+ "id": str(uuid4()),
+ "fire_id": h["fire_id"],
+ "fire_lat": h["fire_lat"],
+ "fire_lon": h["fire_lon"],
+ "aircraft_hex": h["aircraft_hex"],
+ "aircraft_type": h["aircraft_type"],
+ "aircraft_lat": h["aircraft_lat"],
+ "aircraft_lon": h["aircraft_lon"],
+ "distance_mi": h["distance_mi"],
+ "seen_at": ts.replace(microsecond=0),
+ },
+ )
+ n += 1
+ except Exception:
+ continue
+ try:
+ await session.commit()
+ except Exception:
+ return 0
+ return n
+
+
+async def recent_hits(limit: int = 200) -> list[dict]:
+ try:
+ async with async_session() as session:
+ rows = (await session.execute(
+ text(
+ """
+ SELECT fire_id, fire_lat, fire_lon,
+ aircraft_hex, aircraft_type, aircraft_lat, aircraft_lon,
+ distance_mi, seen_at
+ FROM fire_aircraft_hits
+ ORDER BY seen_at DESC
+ LIMIT :limit
+ """
+ ),
+ {"limit": limit},
+ )).mappings().all()
+ out = []
+ for r in rows:
+ item = dict(r)
+ if item.get("seen_at") is not None:
+ item["seen_at"] = item["seen_at"].isoformat()
+ out.append(item)
+ return out
+ except Exception:
+ return []
+
+
+async def correlate_and_notify(fires: list[dict], aircraft: list[dict]) -> list[dict]:
+ hits = correlate_aircraft_to_fires(fires, aircraft)
+ if not hits:
+ return []
+ now = datetime.now(timezone.utc)
+ fresh = []
+ for h in hits:
+ key = (h["fire_id"], h["aircraft_hex"])
+ prev = _recent.get(key)
+ if prev is not None and now - prev < _COOLDOWN:
+ continue
+ _recent[key] = now
+ fresh.append(h)
+ if not fresh:
+ return []
+ await persist_hits(fresh, seen_at=now)
+ from ws_manager import manager
+ for h in fresh:
+ body = {**h, "seen_at": now.isoformat()}
+ await manager.publish_point(
+ "fire_aircraft", body, lat=h["aircraft_lat"], lon=h["aircraft_lon"],
+ )
+ await manager.publish_point(
+ "fire_aircraft", body, lat=h["fire_lat"], lon=h["fire_lon"],
+ )
+ return fresh
diff --git a/app/fire_sources.py b/app/fire_sources.py
index 838d974..5faedd2 100644
--- a/app/fire_sources.py
+++ b/app/fire_sources.py
@@ -40,6 +40,7 @@ from config import (
NATS_URL,
)
from keystore import get_api_key
+from upstream_cache import firms_cache
logger = logging.getLogger("osint.firms")
@@ -185,12 +186,16 @@ async def ingest_fires(bbox: str | None = None) -> int:
total_published = 0
async with httpx.AsyncClient(timeout=FIRMS_TIMEOUT) as client:
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
+ cache_key = (dataset, area, FIRMS_DAYS)
+ text = firms_cache.get(cache_key)
+ if text is None:
+ 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_cache[cache_key] = 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]:
diff --git a/app/geofence.py b/app/geofence.py
new file mode 100644
index 0000000..398c73b
--- /dev/null
+++ b/app/geofence.py
@@ -0,0 +1,321 @@
+"""Geofences: GeoJSON polygons, ST_Intersects on ingest, WS alerts.
+
+``/api/alerts`` is the dashboard entity/keyword table — geofence hits live
+in ``geofence_alerts`` and fan out as WS type ``geofence_alert``.
+"""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime, timedelta, timezone
+from typing import Any
+from uuid import uuid4
+
+from sqlalchemy import text
+
+from database import async_session
+
+
+def _rings_from_geojson(geojson: dict) -> list[list[list[float]]]:
+ if not isinstance(geojson, dict):
+ raise ValueError("geojson must be an object")
+ gj = geojson
+ if gj.get("type") == "Feature":
+ gj = gj.get("geometry") or {}
+ if gj.get("type") == "FeatureCollection":
+ raise ValueError("FeatureCollection is not a single polygon")
+ if gj.get("type") != "Polygon":
+ raise ValueError("geojson must be a Polygon")
+ coords = gj.get("coordinates")
+ if not isinstance(coords, list) or not coords:
+ raise ValueError("polygon has no rings")
+ rings: list[list[list[float]]] = []
+ for ring in coords:
+ if not isinstance(ring, list) or len(ring) < 4:
+ raise ValueError("polygon ring needs ≥4 positions (closed)")
+ pts = []
+ for pt in ring:
+ if not isinstance(pt, (list, tuple)) or len(pt) < 2:
+ raise ValueError("position must be [lon, lat]")
+ pts.append([float(pt[0]), float(pt[1])])
+ rings.append(pts)
+ return rings
+
+
+def validate_polygon_geojson(geojson: dict) -> dict:
+ """Return a canonical Polygon GeoJSON or raise ValueError."""
+ rings = _rings_from_geojson(geojson)
+ return {"type": "Polygon", "coordinates": rings}
+
+
+def _ring_contains(lon: float, lat: float, ring: list[list[float]]) -> bool:
+ """Ray-cast even-odd rule. Ring is [lon, lat] positions."""
+ inside = False
+ n = len(ring)
+ if n < 4:
+ return False
+ j = n - 1
+ for i in range(n):
+ xi, yi = ring[i][0], ring[i][1]
+ xj, yj = ring[j][0], ring[j][1]
+ intersects = ((yi > lat) != (yj > lat)) and (
+ lon < (xj - xi) * (lat - yi) / ((yj - yi) or 1e-16) + xi
+ )
+ if intersects:
+ inside = not inside
+ j = i
+ return inside
+
+
+def point_in_geojson(lon: float, lat: float, geojson: dict) -> bool:
+ """True if (lon, lat) is inside the outer ring and outside holes."""
+ try:
+ rings = _rings_from_geojson(geojson)
+ except (ValueError, TypeError, KeyError):
+ return False
+ if not _ring_contains(lon, lat, rings[0]):
+ return False
+ for hole in rings[1:]:
+ if _ring_contains(lon, lat, hole):
+ return False
+ return True
+
+
+def matching_geofences(lon: float, lat: float, fences: list[dict]) -> list[dict]:
+ hits = []
+ for fence in fences:
+ if not fence.get("active", True):
+ continue
+ gj = fence.get("geojson") or {}
+ if point_in_geojson(lon, lat, gj):
+ hits.append(fence)
+ return hits
+
+
+# In-process copy of active fences so ingest does not round-trip Postgres
+# on every AIS frame. CRUD endpoints refresh this list.
+_cache: list[dict] = []
+_recent_hits: dict[tuple[str, str], datetime] = {}
+_HIT_COOLDOWN = timedelta(minutes=5)
+
+
+async def refresh_cache() -> list[dict]:
+ global _cache
+ async with async_session() as session:
+ rows = (await session.execute(text(
+ "SELECT id::text, name, geojson, active FROM geofences"
+ ))).mappings().all()
+ _cache = [
+ {
+ "id": r["id"],
+ "name": r["name"],
+ "geojson": r["geojson"] if isinstance(r["geojson"], dict)
+ else json.loads(r["geojson"] or "{}"),
+ "active": bool(r["active"]),
+ }
+ for r in rows
+ ]
+ return _cache
+
+
+def cached_fences() -> list[dict]:
+ return list(_cache)
+
+
+async def list_geofences() -> list[dict]:
+ if not _cache:
+ try:
+ await refresh_cache()
+ except Exception:
+ return []
+ return cached_fences()
+
+
+async def create_geofence(name: str, geojson: dict, active: bool = True) -> dict:
+ polygon = validate_polygon_geojson(geojson)
+ gid = str(uuid4())
+ gj = json.dumps(polygon)
+ async with async_session() as session:
+ await session.execute(
+ text(
+ """
+ INSERT INTO geofences (id, name, geojson, geom, active)
+ VALUES (
+ :id, :name, CAST(:geojson AS jsonb),
+ ST_SetSRID(ST_GeomFromGeoJSON(:geojson), 4326),
+ :active
+ )
+ """
+ ),
+ {"id": gid, "name": name, "geojson": gj, "active": 1 if active else 0},
+ )
+ await session.commit()
+ row = {"id": gid, "name": name, "geojson": polygon, "active": active}
+ _cache.append(row)
+ return row
+
+
+async def update_geofence(gid: str, *, name: str | None = None,
+ geojson: dict | None = None,
+ active: bool | None = None) -> dict | None:
+ current = next((f for f in _cache if f["id"] == gid), None)
+ if current is None:
+ await refresh_cache()
+ current = next((f for f in _cache if f["id"] == gid), None)
+ if current is None:
+ return None
+ if name is not None:
+ current["name"] = name
+ if geojson is not None:
+ current["geojson"] = validate_polygon_geojson(geojson)
+ if active is not None:
+ current["active"] = active
+ gj = json.dumps(current["geojson"])
+ async with async_session() as session:
+ await session.execute(
+ text(
+ """
+ UPDATE geofences SET
+ name = :name,
+ geojson = CAST(:geojson AS jsonb),
+ geom = ST_SetSRID(ST_GeomFromGeoJSON(:geojson), 4326),
+ active = :active,
+ updated_at = now()
+ WHERE id = CAST(:id AS uuid)
+ """
+ ),
+ {
+ "id": gid,
+ "name": current["name"],
+ "geojson": gj,
+ "active": 1 if current["active"] else 0,
+ },
+ )
+ await session.commit()
+ return current
+
+
+async def delete_geofence(gid: str) -> bool:
+ async with async_session() as session:
+ result = await session.execute(
+ text("DELETE FROM geofences WHERE id = CAST(:id AS uuid)"),
+ {"id": gid},
+ )
+ await session.commit()
+ _cache[:] = [f for f in _cache if f["id"] != gid]
+ return bool(result.rowcount)
+
+
+async def st_intersects(lon: float, lat: float) -> list[dict]:
+ """PostGIS ST_Intersects against active geofences.
+
+ Falls back to the in-memory GeoJSON test if the DB is unreachable so
+ ingest never dies because a fence check failed.
+ """
+ try:
+ async with async_session() as session:
+ rows = (await session.execute(
+ text(
+ """
+ SELECT id::text, name, geojson, active
+ FROM geofences
+ WHERE active = 1
+ AND ST_Intersects(
+ geom,
+ ST_SetSRID(ST_MakePoint(:lon, :lat), 4326)
+ )
+ """
+ ),
+ {"lon": lon, "lat": lat},
+ )).mappings().all()
+ return [
+ {
+ "id": r["id"],
+ "name": r["name"],
+ "geojson": r["geojson"] if isinstance(r["geojson"], dict)
+ else json.loads(r["geojson"] or "{}"),
+ "active": True,
+ }
+ for r in rows
+ ]
+ except Exception:
+ return matching_geofences(lon, lat, cached_fences())
+
+
+async def record_and_notify(
+ *,
+ source_kind: str,
+ entity_id: str,
+ lat: float,
+ lon: float,
+ payload: dict[str, Any] | None = None,
+) -> int:
+ """Insert a geofence_alerts row per hit and WS-push to viewport clients.
+
+ PostGIS ST_Intersects is the source of truth. The in-process GeoJSON
+ cache is not a reject filter — the FIRMS ingester never fills it.
+ """
+ hits = await st_intersects(lon, lat)
+ if not hits:
+ return 0
+ from ws_manager import manager
+
+ sent = 0
+ now = datetime.now(timezone.utc)
+ fresh = []
+ for fence in hits:
+ key = (str(fence["id"]), str(entity_id))
+ prev = _recent_hits.get(key)
+ if prev is not None and now - prev < _HIT_COOLDOWN:
+ continue
+ _recent_hits[key] = now
+ fresh.append(fence)
+ if not fresh:
+ return 0
+ hits = fresh
+ async with async_session() as session:
+ for fence in hits:
+ aid = str(uuid4())
+ body = {
+ "id": aid,
+ "geofence_id": fence["id"],
+ "geofence_name": fence.get("name"),
+ "source_kind": source_kind,
+ "entity_id": str(entity_id),
+ "lat": lat,
+ "lon": lon,
+ "payload": payload or {},
+ "created_at": now.isoformat(),
+ }
+ try:
+ await session.execute(
+ text(
+ """
+ INSERT INTO geofence_alerts
+ (id, geofence_id, source_kind, entity_id, lat, lon, payload)
+ VALUES (
+ CAST(:id AS uuid), CAST(:geofence_id AS uuid),
+ :source_kind, :entity_id, :lat, :lon, CAST(:payload AS jsonb)
+ )
+ """
+ ),
+ {
+ "id": aid,
+ "geofence_id": fence["id"],
+ "source_kind": source_kind,
+ "entity_id": str(entity_id),
+ "lat": lat,
+ "lon": lon,
+ "payload": json.dumps(payload or {}),
+ },
+ )
+ except Exception:
+ pass
+ sent += await manager.publish_point(
+ "geofence_alert", body, lat=lat, lon=lon,
+ )
+ try:
+ await session.commit()
+ except Exception:
+ pass
+ return sent
diff --git a/app/ingestor.py b/app/ingestor.py
index 56909b5..54d49cb 100644
--- a/app/ingestor.py
+++ b/app/ingestor.py
@@ -95,6 +95,23 @@ async def ingest_fire_row(msg: dict) -> bool:
"ingested fire %.5f,%.5f %s satellite=%s",
row["latitude"], row["longitude"], row["acq_time"], row["satellite"],
)
+ lat, lon = row["latitude"], row["longitude"]
+ from geofence import record_and_notify
+ await record_and_notify(
+ source_kind="firms",
+ entity_id=f"{lat},{lon},{row['satellite']}",
+ lat=lat, lon=lon, payload={"satellite": row["satellite"]},
+ )
+ from live_layers import aircraft_last_known
+ from fire_aircraft import correlate_and_notify
+ from tracks import recent_markers
+ acs = list(aircraft_last_known.values()) or await recent_markers("aircraft")
+ if acs:
+ fire = {
+ "id": f"firms:{lat:.4f},{lon:.4f}",
+ "lat": lat, "lon": lon, "label": "FIRMS",
+ }
+ await correlate_and_notify([fire], acs)
return inserted
diff --git a/app/live_layers.py b/app/live_layers.py
index 5daf056..e5b9f0e 100644
--- a/app/live_layers.py
+++ b/app/live_layers.py
@@ -69,6 +69,9 @@ vessel_lock = asyncio.Lock()
# Viewport-following accumulates vessels across every region visited in a
# session — keep the in-memory store bounded (oldest entries evicted).
_MAX_VESSELS = 6000
+# Last ADS-B snapshot + WFIGS points for fire↔tanker correlation.
+aircraft_last_known: dict[str, dict] = {}
+fire_last_known: list[dict] = []
def overlay_catalog() -> dict:
@@ -576,6 +579,25 @@ async def fetch_aircraft(bbox: str, limit: int = DEFAULT_LIMIT) -> list[dict]:
return transform_adsb_lol(await _get_json(url))
rows = await _ttl_get(cache_key, 8.0, _load)
+ from ws_manager import manager
+ from tracks import record_position
+ from geofence import record_and_notify
+ aircraft_last_known.clear()
+ for m in rows:
+ aircraft_last_known[str(m.get("id"))] = m
+ mlat, mlon = m.get("lat"), m.get("lon")
+ if mlat is None or mlon is None:
+ continue
+ await record_position("aircraft", m)
+ if manager.has_clients():
+ await manager.publish_point("adsb", m, lat=mlat, lon=mlon)
+ await record_and_notify(
+ source_kind="adsb", entity_id=str(m.get("id")),
+ lat=mlat, lon=mlon, payload=m,
+ )
+ if fire_last_known:
+ from fire_aircraft import correlate_and_notify
+ await correlate_and_notify(fire_last_known, rows)
return filter_points_bbox(rows, minlon, minlat, maxlon, maxlat, limit)
@@ -613,7 +635,7 @@ async def upsert_vessel(marker: dict) -> None:
label = marker.get("label")
if not label or label == vid:
label = prev.get("label") or vid
- vessel_last_known[vid] = {
+ stored = {
**to_marker(
vid, lat, lon,
heading=marker.get("heading") if marker.get("heading") is not None else prev.get("heading"),
@@ -623,6 +645,7 @@ async def upsert_vessel(marker: dict) -> None:
),
"seen_at": datetime.now(timezone.utc).isoformat(),
}
+ vessel_last_known[vid] = stored
if len(vessel_last_known) > _MAX_VESSELS:
excess = len(vessel_last_known) - int(_MAX_VESSELS * 0.9)
oldest = sorted(
@@ -631,6 +654,15 @@ async def upsert_vessel(marker: dict) -> None:
)[:excess]
for k in oldest:
vessel_last_known.pop(k, None)
+ if lat is not None and lon is not None:
+ from ws_manager import manager
+ from tracks import record_position
+ from geofence import record_and_notify
+ await manager.publish_point("ais", stored, lat=lat, lon=lon)
+ await record_position("vessel", stored)
+ await record_and_notify(
+ source_kind="ais", entity_id=vid, lat=lat, lon=lon, payload=stored,
+ )
def _wfigs_params(bbox: str | None, *, offset_m: float = 250.0) -> dict:
@@ -662,6 +694,10 @@ async def fetch_fire_incidents(bbox: str | None, limit: int = DEFAULT_LIMIT) ->
return transform_wfigs_incidents(await _get_json(WFIGS_INCIDENTS, params))
rows = await _ttl_get(f"wfigs:inc:{bbox_cell_key(bbox)}", 600.0, _load)
+ fire_last_known[:] = list(rows)
+ if rows and aircraft_last_known:
+ from fire_aircraft import correlate_and_notify
+ await correlate_and_notify(rows, list(aircraft_last_known.values()))
return rows[:limit]
diff --git a/app/main.py b/app/main.py
index 421fbfc..aad0663 100644
--- a/app/main.py
+++ b/app/main.py
@@ -21,7 +21,7 @@ from pathlib import Path
from uuid import UUID
import structlog
-from fastapi import FastAPI, HTTPException, Query
+from fastapi import BackgroundTasks, FastAPI, HTTPException, Query, WebSocket, WebSocketDisconnect
from fastapi.middleware.gzip import GZipMiddleware
from fastapi.responses import FileResponse, HTMLResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
@@ -41,6 +41,7 @@ from schemas import (
KeyOut, KeyValueIn,
SearchResult, SentimentSummary, SourceType,
SearchQuery, TimelinePoint, VesselBboxUpdate,
+ GeofenceCreate, GeofenceUpdate,
)
from ingestor import ingest_event, fetch_and_process
from sources import ingest_rss_feed, ingest_gdelt, ingest_earthquakes, ingest_social_signals
@@ -61,6 +62,11 @@ async def _lifespan(app: FastAPI):
await init_extensions()
from live_layers import close_http, init_http
await init_http()
+ try:
+ from geofence import refresh_cache
+ await refresh_cache()
+ except Exception:
+ pass
from config import AISSTREAM_IN_APP
ais_task = None
if AISSTREAM_IN_APP:
@@ -664,6 +670,56 @@ async def trigger_social_ingest(query: str = "", max_items: int = 50):
return {"status": "ok", "signals_ingested": count}
+@app.post("/api/ingest/masscan")
+async def trigger_masscan(background_tasks: BackgroundTasks):
+ """Queue one masscan pass at ≤200 pps. Does not block the request on the scan."""
+ from bg_jobs import MASSCAN_PPS_CAP, schedule_masscan_pass
+
+ async def _kick() -> None:
+ schedule_masscan_pass()
+
+ background_tasks.add_task(_kick)
+ return JSONResponse(
+ {"status": "accepted", "rate_pps": MASSCAN_PPS_CAP},
+ status_code=202,
+ )
+
+
+@app.websocket("/ws/live")
+async def live_ws(ws: WebSocket):
+ """Viewport-filtered AIS/ADS-B fan-out. Client sends {type:viewport,bbox}."""
+ from ws_manager import manager
+
+ client_id = str(id(ws))
+ await ws.accept()
+ queue = manager.register(client_id)
+
+ async def _pump() -> None:
+ try:
+ while True:
+ msg = await queue.get()
+ await ws.send_json(msg)
+ except Exception: # noqa: BLE001
+ return
+
+ pump = asyncio.create_task(_pump())
+ try:
+ while True:
+ data = await ws.receive_json()
+ if not isinstance(data, dict):
+ continue
+ if data.get("type") == "viewport" and data.get("bbox"):
+ try:
+ manager.set_viewport(client_id, parse_bbox(str(data["bbox"])))
+ except ValueError:
+ continue
+ except WebSocketDisconnect:
+ pass
+ finally:
+ pump.cancel()
+ manager.unregister(client_id)
+
+
@app.post("/api/ingest/process")
async def trigger_nats_processing(batch_size: int = 100):
"""Process pending NATS JetStream messages."""
@@ -1135,10 +1191,15 @@ async def map_radar():
async def list_aircraft(
bbox: str = Query(..., description="minlon,minlat,maxlon,maxlat"),
limit: int = Query(2000, ge=1, le=5000),
+ timestamp: str | None = Query(None, description="ISO time — DVR 1-min tracks instead of live"),
):
"""Viewport ADS-B last-known (ADSB.lol). Requires bbox; radius clamped ≤ 150 nm."""
_parse_bbox_query(bbox)
try:
+ from tracks import fetch_positions_at, parse_timestamp
+ ts = parse_timestamp(timestamp)
+ if ts is not None:
+ return overlay_json(await fetch_positions_at("aircraft", ts, bbox, limit), 5)
return overlay_json(await fetch_aircraft(bbox, limit), 5)
except ValueError as exc:
raise HTTPException(422, str(exc)) from exc
@@ -1166,11 +1227,16 @@ async def list_trains(
async def list_vessels(
bbox: str | None = Query(None, description="minlon,minlat,maxlon,maxlat"),
limit: int = Query(2000, ge=1, le=5000),
+ timestamp: str | None = Query(None, description="ISO time — DVR 1-min tracks instead of live"),
):
"""AIS last-known from the server-side AISStream worker. Empty without a key."""
if bbox:
_parse_bbox_query(bbox)
try:
+ from tracks import fetch_positions_at, parse_timestamp
+ ts = parse_timestamp(timestamp)
+ if ts is not None:
+ return overlay_json(await fetch_positions_at("vessel", ts, bbox, limit), 5)
return overlay_json(await fetch_vessels(bbox, limit), 5)
except ValueError as exc:
raise HTTPException(422, str(exc)) from exc
@@ -1204,6 +1270,90 @@ async def vessels_subscribe(payload: VesselBboxUpdate):
return {"ok": True, "bbox": raw}
+@app.get("/api/tracks/range")
+async def tracks_range():
+ """Earliest/latest 1-minute track buckets for the DVR slider."""
+ from tracks import track_range
+ return await track_range()
+
+
+@app.get("/api/geofences")
+async def api_list_geofences():
+ """Drawn GeoJSON polygons. Not /api/alerts (entity/keyword)."""
+ from geofence import list_geofences
+ return await list_geofences()
+
+
+@app.post("/api/geofences", status_code=201)
+async def api_create_geofence(payload: GeofenceCreate):
+ from geofence import create_geofence, validate_polygon_geojson
+ try:
+ validate_polygon_geojson(payload.geojson)
+ except ValueError as exc:
+ raise HTTPException(422, str(exc)) from exc
+ try:
+ return await create_geofence(payload.name, payload.geojson, payload.active)
+ except HTTPException:
+ raise
+ except Exception as exc:
+ raise HTTPException(503, f"geofence persist failed: {exc}") from exc
+
+
+@app.patch("/api/geofences/{gid}")
+async def api_update_geofence(gid: str, payload: GeofenceUpdate):
+ from geofence import update_geofence, validate_polygon_geojson
+ if payload.geojson is not None:
+ try:
+ validate_polygon_geojson(payload.geojson)
+ except ValueError as exc:
+ raise HTTPException(422, str(exc)) from exc
+ row = await update_geofence(
+ gid, name=payload.name, geojson=payload.geojson, active=payload.active,
+ )
+ if row is None:
+ raise HTTPException(404, "geofence not found")
+ return row
+
+
+@app.delete("/api/geofences/{gid}", status_code=204)
+async def api_delete_geofence(gid: str):
+ from geofence import delete_geofence
+ await delete_geofence(gid)
+ return None
+
+
+@app.get("/api/geofence-alerts")
+async def api_geofence_alerts(limit: int = Query(100, ge=1, le=500)):
+ from sqlalchemy import text as sql_text
+ try:
+ async with async_session() as session:
+ rows = (await session.execute(sql_text(
+ """
+ SELECT id::text, geofence_id::text, source_kind, entity_id,
+ lat, lon, payload, created_at
+ FROM geofence_alerts
+ ORDER BY created_at DESC
+ LIMIT :limit
+ """
+ ), {"limit": limit})).mappings().all()
+ out = []
+ for r in rows:
+ item = dict(r)
+ if item.get("created_at") is not None:
+ item["created_at"] = item["created_at"].isoformat()
+ out.append(item)
+ return out
+ except Exception:
+ return []
+
+
+@app.get("/api/fire-aircraft")
+async def api_fire_aircraft(limit: int = Query(200, ge=1, le=1000)):
+ """Persisted firefighting ADS-B × wildfire correlations (20 mi)."""
+ from fire_aircraft import recent_hits
+ return await recent_hits(limit)
+
+
@app.get("/api/fire-incidents")
async def list_fire_incidents(
bbox: str | None = Query(None),
diff --git a/app/requirements.txt b/app/requirements.txt
index 508bcb6..691ab71 100644
--- a/app/requirements.txt
+++ b/app/requirements.txt
@@ -11,3 +11,4 @@ feedparser>=6.0
python-dateutil>=2.9
structlog>=24.4
websockets>=14
+cachetools>=5.5
diff --git a/app/run_ingester.py b/app/run_ingester.py
index 56e0062..e2f0c18 100644
--- a/app/run_ingester.py
+++ b/app/run_ingester.py
@@ -107,6 +107,11 @@ async def main() -> None:
"ingester starting (rss=%d feeds, gdelt_q=%r, quakes=%s, fires=%s, interval=%ss)",
len(RSS_URLS), GDELT_QUERY, ENABLE_QUAKES, ENABLE_FIRES, INTERVAL,
)
+ try:
+ from geofence import refresh_cache
+ await refresh_cache()
+ except Exception:
+ logger.exception("geofence cache refresh failed (ST_Intersects still runs on ingest)")
tasks: list[asyncio.Task] = []
if ENABLE_FIRES:
# Fire ingest only starts once FIRMS_MAP_KEY is set (ingest_fires logs
diff --git a/app/schemas.py b/app/schemas.py
index f49d3a7..e27b002 100644
--- a/app/schemas.py
+++ b/app/schemas.py
@@ -310,3 +310,15 @@ class VesselBboxUpdate(BaseModel):
bbox: str | None = None
+
+class GeofenceCreate(BaseModel):
+ name: str
+ geojson: dict
+ active: bool = True
+
+
+class GeofenceUpdate(BaseModel):
+ name: Optional[str] = None
+ geojson: Optional[dict] = None
+ active: Optional[bool] = None
+
diff --git a/app/sources.py b/app/sources.py
index 3d76b34..804dea1 100644
--- a/app/sources.py
+++ b/app/sources.py
@@ -12,6 +12,7 @@ import feedparser
import nats
from config import NATS_URL
+from upstream_cache import rss_cache
logger = logging.getLogger("osint.sources")
@@ -40,12 +41,16 @@ async def publish_event(subject: str, event: dict):
# ─── RSS Feed Ingestor ──────────────────────────────────────────────────
-async def ingest_rss_feed(feed_url: str):
+async def ingest_rss_feed(feed_url: str, source_id: str | None = None):
"""Fetch and parse an RSS feed, publish items to NATS."""
- async with httpx.AsyncClient(timeout=30) as client:
- resp = await client.get(feed_url)
- resp.raise_for_status()
- feed = feedparser.parse(resp.text)
+ text = rss_cache.get(feed_url)
+ if text is None:
+ async with httpx.AsyncClient(timeout=30) as client:
+ resp = await client.get(feed_url)
+ resp.raise_for_status()
+ text = resp.text
+ rss_cache[feed_url] = text
+ feed = feedparser.parse(text)
count = 0
for entry in feed.entries[:100]: # max 100 per run
diff --git a/app/static/index.html b/app/static/index.html
index 0d8e730..d07f1ec 100644
--- a/app/static/index.html
+++ b/app/static/index.html
@@ -265,6 +265,26 @@
.hud-chip .c.cams { background: #4ade80; box-shadow: 0 0 6px #4ade80; }
.hud-chip .c.blips { background: var(--magenta); box-shadow: 0 0 6px var(--magenta); }
.hud-chip.off { opacity: 0.4; }
+ #dvr-bar {
+ position: absolute; left: 50%; bottom: 12px; transform: translateX(-50%);
+ z-index: 520; display: flex; align-items: center; gap: 0.55rem;
+ background: rgba(6,11,20,0.86); border: 1px solid var(--line-hi);
+ border-radius: 4px; padding: 0.28rem 0.7rem;
+ font-family: 'Share Tech Mono', monospace; font-size: 0.66rem; color: var(--muted);
+ min-width: 280px; max-width: 52%;
+ }
+ #dvr-bar input[type=range] { flex: 1; accent-color: var(--cyan); }
+ #dvr-bar.live #dvr-live { color: var(--green); text-shadow: 0 0 8px rgba(83,240,165,0.5); }
+ #dvr-toast {
+ position: absolute; top: 12px; right: 12px; z-index: 700;
+ max-width: 280px; pointer-events: none;
+ }
+ #dvr-toast .gf-alert {
+ background: rgba(255,46,151,0.16); border: 1px solid var(--magenta);
+ color: var(--text); font-family: 'Share Tech Mono', monospace;
+ font-size: 0.68rem; padding: 0.4rem 0.6rem; margin-bottom: 0.35rem;
+ border-radius: 4px;
+ }
/* ── Layer control panel (dark HUD) ── */
#layer-panel {
@@ -754,6 +774,12 @@
Needs AISSTREAM_API_KEY in Keys.
+
+
+
+
+
No Leaflet.Draw — click vertices, double-click to close. Saved to POST /api/geofences.
+
@@ -776,6 +802,12 @@
CAMS —
BLIPS —
+
+
+
+ live
+
+
@@ -1658,6 +1690,76 @@ function bboxCell() {
if (!map) return '';
return currentBBox().split(',').map(n => Number(n).toFixed(2)).join(',') + '@' + map.getZoom();
}
+let liveWs = null;
+const FF_ICAO = new Set(['AT802','AT8T','AT8P','C130','C30J','C130J','DC10','MD10','MD87','S64','UH60','S70','CL415','CL215','B350','OV10','C208']);
+let firefighterHex = new Set();
+let dvrTs = null;
+let dvrMinMs = 0, dvrMaxMs = 0;
+let gfDrawOn = false, gfVerts = [], gfLayer = null, gfSaved = null;
+function liveWsUrl() {
+ const proto = location.protocol === 'https:' ? 'wss:' : 'ws:';
+ return proto + '//' + location.host + '/ws/live';
+}
+function sendLiveViewport() {
+ if (!liveWs || liveWs.readyState !== 1 || !map) return;
+ liveWs.send(JSON.stringify({ type: 'viewport', bbox: currentBBox() }));
+}
+function upsertLivePoint(group, p, colorFn, feed) {
+ if (!map || !p || p.id == null || p.lat == null || p.lon == null) return group;
+ if (!group || !map.hasLayer(group) || !group._osintById) {
+ return renderPoints(group, [p], colorFn, true, feed);
+ }
+ const id = String(p.id);
+ const col = sanitizeColor(colorFn(p), '#35e0ff');
+ let m = group._osintById.get(id);
+ if (m) {
+ m.setLatLng([p.lat, p.lon]);
+ if (feed) m.setIcon(feedIcon(feed, col, p.heading));
+ else if (m.setStyle) m.setStyle({ color: col, fillColor: col });
+ } else {
+ m = makePointMarker(p, colorFn, feed, pointCanvas());
+ if (m) { group.addLayer(m); group._osintById.set(id, m); }
+ }
+ return group;
+}
+function applyLiveMarker(kind, p) {
+ if (!p) return;
+ if (kind === 'ais' && vesselsOn && map && map.getZoom() > 3 && !dvrTs) {
+ vesselsGroup = upsertLivePoint(vesselsGroup, p, q => {
+ const sog = Number(q.speed || 0);
+ return sog > 0.5 ? '#2dd4bf' : '#64748b';
+ }, 'vessel');
+ } else if (kind === 'adsb' && acOn && map && map.getZoom() > 3 && !dvrTs) {
+ acGroup = upsertLivePoint(acGroup, p, q => acColor(q), 'ac');
+ } else if (kind === 'geofence_alert') {
+ showGeofenceToast(p);
+ } else if (kind === 'fire_aircraft') {
+ firefighterHex.add(String(p.aircraft_hex || ''));
+ if (acOn && p.aircraft_lat != null) {
+ acGroup = upsertLivePoint(acGroup, {
+ id: p.aircraft_hex, lat: p.aircraft_lat, lon: p.aircraft_lon,
+ label: p.label || p.aircraft_hex, extra: { type: p.aircraft_type, firefighter: true },
+ }, q => acColor(q), 'ac');
+ }
+ }
+}
+function acColor(p) {
+ const t = String(((p.extra || {}).type || '')).toUpperCase();
+ if ((p.extra || {}).firefighter || firefighterHex.has(String(p.id)) || FF_ICAO.has(t)) return '#fb923c';
+ return altColor((p.extra || {}).alt_baro);
+}
+function connectLiveWs() {
+ if (liveWs && (liveWs.readyState === 0 || liveWs.readyState === 1)) return;
+ try { liveWs = new WebSocket(liveWsUrl()); } catch (e) { return; }
+ liveWs.onopen = () => sendLiveViewport();
+ liveWs.onmessage = (ev) => {
+ try {
+ const msg = JSON.parse(ev.data);
+ applyLiveMarker(msg.type, msg.payload);
+ } catch (e) { /* ignore malformed */ }
+ };
+ liveWs.onclose = () => { liveWs = null; setTimeout(connectLiveWs, 4000); };
+}
function overlayFetch(url) {
return fetch(url, overlayAbort ? { signal: overlayAbort.signal } : {});
}
@@ -1754,7 +1856,8 @@ async function initMap() {
if (camsOn) loadCams();
if (blipsOn) loadBlips();
refreshLiveOverlays();
- }, 300);
+ sendLiveViewport();
+ }, 500);
});
// Static Blue Marble has no time-domain fetch — don't block overlays on GIBS.
mapLayerChanged();
@@ -1778,6 +1881,7 @@ async function initMap() {
if (camsOn) loadCams();
if (blipsOn) loadBlips();
refreshLiveOverlays();
+ connectLiveWs();
} catch(e) {
hint.textContent = `Failed to load map layers: ${e.message || e}`;
console.error('Map init failed', e);
@@ -2226,6 +2330,8 @@ function refreshLiveOverlays() {
if (trainsOn) loadTrains();
if (vesselsOn) loadVessels();
if (stormsOn) loadStorms();
+ loadGeofences();
+ loadFireAircraftHits();
}
function addExtraAttrib(html) {
if (!map || !html || extraAttribs.has(html)) return;
@@ -2554,10 +2660,10 @@ async function loadAircraft() {
}
const req = ++overlayReq.ac;
try {
- const r = await overlayFetch(`${API}/api/aircraft?bbox=${currentBBox()}`);
+ const r = await overlayFetch(`${API}/api/aircraft?bbox=${currentBBox()}${dvrQs()}`);
const pts = await r.json();
if (req !== overlayReq.ac) return;
- acGroup = renderPoints(acGroup, Array.isArray(pts) ? pts : [], p => altColor((p.extra || {}).alt_baro), true, 'ac');
+ acGroup = renderPoints(acGroup, Array.isArray(pts) ? pts : [], p => acColor(p), true, 'ac');
document.getElementById('lp-ac-count').textContent = (pts.length || 0).toLocaleString();
addExtraAttrib('ADSB.lol ODbL');
} catch (e) {
@@ -2622,7 +2728,7 @@ async function loadVessels() {
}).catch(() => {});
}
try {
- const r = await overlayFetch(`${API}/api/vessels?bbox=${bb}`);
+ const r = await overlayFetch(`${API}/api/vessels?bbox=${bb}${dvrQs()}`);
const pts = await r.json();
if (req !== overlayReq.vessels) return;
vesselsGroup = renderPoints(vesselsGroup, Array.isArray(pts) ? pts : [], p => {
@@ -2674,6 +2780,122 @@ initMap();
// contend with the first bbox burst. Payload is slim (no article bodies).
setTimeout(() => loadNews(true), 400);
setInterval(checkHealth, 30000);
+
+/* Phase 2: DVR slider + geofence draw (no Leaflet.Draw) + fire/aircraft hits */
+function dvrQs() { return dvrTs ? `×tamp=${encodeURIComponent(dvrTs)}` : ''; }
+function dvrGoLive() {
+ dvrTs = null;
+ const bar = document.getElementById('dvr-bar');
+ if (bar) bar.classList.add('live');
+ const sl = document.getElementById('dvr-slider');
+ if (sl) sl.value = sl.max;
+ const lab = document.getElementById('dvr-label');
+ if (lab) lab.textContent = 'live';
+ if (acOn) loadAircraft();
+ if (vesselsOn) loadVessels();
+}
+function dvrScrub(v) {
+ if (!dvrMaxMs) { dvrGoLive(); return; }
+ const ms = dvrMinMs + (Number(v) / 360) * (dvrMaxMs - dvrMinMs);
+ dvrTs = new Date(ms).toISOString();
+ const bar = document.getElementById('dvr-bar');
+ if (bar) bar.classList.toggle('live', Number(v) >= 359);
+ if (Number(v) >= 359) { dvrGoLive(); return; }
+ const lab = document.getElementById('dvr-label');
+ if (lab) lab.textContent = dvrTs.slice(11, 16) + 'Z';
+ if (acOn) loadAircraft();
+ if (vesselsOn) loadVessels();
+}
+async function initDvrRange() {
+ try {
+ const r = await fetch(`${API}/api/tracks/range`);
+ const b = await r.json();
+ dvrMinMs = Date.parse(b.min);
+ dvrMaxMs = Date.parse(b.max);
+ if (!dvrMinMs || !dvrMaxMs) {
+ dvrMaxMs = Date.now();
+ dvrMinMs = dvrMaxMs - 6 * 3600 * 1000;
+ }
+ } catch (e) {
+ dvrMaxMs = Date.now();
+ dvrMinMs = dvrMaxMs - 6 * 3600 * 1000;
+ }
+}
+function showGeofenceToast(p) {
+ const box = document.getElementById('dvr-toast');
+ if (!box) return;
+ const el = document.createElement('div');
+ el.className = 'gf-alert';
+ el.textContent = `GEOFENCE ${p.geofence_name || ''} · ${p.source_kind} ${p.entity_id || ''}`;
+ box.prepend(el);
+ setTimeout(() => el.remove(), 8000);
+}
+function toggleGeofenceDraw() {
+ gfDrawOn = !gfDrawOn;
+ gfVerts = [];
+ if (gfLayer && map) { map.removeLayer(gfLayer); gfLayer = null; }
+ const btn = document.getElementById('gf-draw');
+ if (btn) btn.textContent = gfDrawOn ? 'Click map… (dbl-click close)' : 'Draw geofence';
+ if (!map) return;
+ if (gfDrawOn) {
+ map.getContainer().style.cursor = 'crosshair';
+ map.on('click', onGfClick);
+ map.on('dblclick', onGfClose);
+ } else {
+ map.getContainer().style.cursor = '';
+ map.off('click', onGfClick);
+ map.off('dblclick', onGfClose);
+ }
+}
+function onGfClick(e) {
+ if (!gfDrawOn) return;
+ L.DomEvent.stop(e);
+ gfVerts.push([e.latlng.lng, e.latlng.lat]);
+ if (gfLayer && map) map.removeLayer(gfLayer);
+ if (gfVerts.length >= 2) {
+ gfLayer = L.polyline(gfVerts.map(v => [v[1], v[0]]), { color: '#ff2e97', weight: 2 }).addTo(map);
+ }
+}
+async function onGfClose(e) {
+ if (!gfDrawOn) return;
+ L.DomEvent.stop(e);
+ if (gfVerts.length < 3) return;
+ const ring = gfVerts.concat([gfVerts[0]]);
+ const geojson = { type: 'Polygon', coordinates: [ring] };
+ try {
+ await fetch(`${API}/api/geofences`, {
+ method: 'POST', headers: { 'Content-Type': 'application/json' },
+ body: JSON.stringify({ name: 'Fence ' + new Date().toISOString().slice(11, 19), geojson }),
+ });
+ } catch (err) { console.error('geofence save failed', err); }
+ toggleGeofenceDraw();
+ loadGeofences();
+}
+async function loadGeofences() {
+ if (!map) return;
+ try {
+ const r = await overlayFetch(`${API}/api/geofences`);
+ const rows = await r.json();
+ if (gfSaved && map.hasLayer(gfSaved)) map.removeLayer(gfSaved);
+ const feats = (Array.isArray(rows) ? rows : []).map(f => ({
+ type: 'Feature', properties: { name: f.name, id: f.id }, geometry: f.geojson,
+ }));
+ gfSaved = L.geoJSON({ type: 'FeatureCollection', features: feats }, {
+ style: { color: '#ff2e97', weight: 2, fillOpacity: 0.08 },
+ }).addTo(map);
+ } catch (e) { if (!isAbort(e)) console.error('geofences load failed', e); }
+}
+async function loadFireAircraftHits() {
+ try {
+ const r = await fetch(`${API}/api/fire-aircraft`);
+ const rows = await r.json();
+ (Array.isArray(rows) ? rows : []).forEach(h => {
+ if (h.aircraft_hex) firefighterHex.add(String(h.aircraft_hex));
+ });
+ } catch (e) { /* empty without table */ }
+}
+initDvrRange();
+