From 5ea9a4e879397013ef627f9cf8188080d79774aa Mon Sep 17 00:00:00 2001 From: Sirius DevOps Date: Fri, 28 Aug 2026 09:33:19 -0400 Subject: [PATCH] feat: lean-Pi WS fan-out, geofences, DVR playback, fire/aircraft correlation Phase 1: in-memory ConnectionManager viewport fan-out, 500ms map debounce, cachetools TTLCache, background masscan/ffmpeg, compose memory caps. Phase 2: PostGIS geofences + ST_Intersects alerts, Timescale 1-min CAGGs and timestamp playback, FIRMS/WFIGS x firefighting ADS-B within 20 miles. No Redis/Kafka/Celery. --- alembic/versions/005_phase2.py | 196 +++++++++++++++++++ app/bg_jobs.py | 103 ++++++++++ app/camera_preview.py | 43 ++--- app/fire_aircraft.py | 186 ++++++++++++++++++ app/fire_sources.py | 17 +- app/geofence.py | 321 ++++++++++++++++++++++++++++++++ app/ingestor.py | 17 ++ app/live_layers.py | 38 +++- app/main.py | 152 ++++++++++++++- app/requirements.txt | 1 + app/run_ingester.py | 5 + app/schemas.py | 12 ++ app/sources.py | 15 +- app/static/index.html | 230 ++++++++++++++++++++++- app/tracks.py | 245 ++++++++++++++++++++++++ app/upstream_cache.py | 11 ++ app/ws_manager.py | 80 ++++++++ docker-compose.yml | 15 +- tests/test_bg_jobs.py | 64 +++++++ tests/test_fire_aircraft.py | 66 +++++++ tests/test_fire_ingest.py | 84 +++++++++ tests/test_fire_sources.py | 2 + tests/test_geofence.py | 136 ++++++++++++++ tests/test_phase1_guardrails.py | 33 ++++ tests/test_phase2_api.py | 51 +++++ tests/test_tracks.py | 50 +++++ tests/test_upstream_cache.py | 104 +++++++++++ tests/test_ws_manager.py | 91 +++++++++ 28 files changed, 2318 insertions(+), 50 deletions(-) create mode 100644 alembic/versions/005_phase2.py create mode 100644 app/bg_jobs.py create mode 100644 app/fire_aircraft.py create mode 100644 app/geofence.py create mode 100644 app/tracks.py create mode 100644 app/upstream_cache.py create mode 100644 app/ws_manager.py create mode 100644 tests/test_bg_jobs.py create mode 100644 tests/test_fire_aircraft.py create mode 100644 tests/test_geofence.py create mode 100644 tests/test_phase1_guardrails.py create mode 100644 tests/test_phase2_api.py create mode 100644 tests/test_tracks.py create mode 100644 tests/test_upstream_cache.py create mode 100644 tests/test_ws_manager.py 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(); + diff --git a/app/tracks.py b/app/tracks.py new file mode 100644 index 0000000..b3a3d83 --- /dev/null +++ b/app/tracks.py @@ -0,0 +1,245 @@ +"""Timescale 1-minute track rollups for DVR playback. + +Live overlays stay in memory. Historical `?timestamp=` reads the 1-minute +continuous aggregates (or an in-process downsample when the DB is down). +""" + +from __future__ import annotations + +import json +from datetime import datetime, timedelta, timezone +from typing import Any, Literal + +from sqlalchemy import text + +from database import async_session +from live_layers import parse_bbox, to_marker + + +TRACK_BUCKET = "1 minute" +Kind = Literal["vessel", "aircraft"] + +_RAW_TABLE = { + "vessel": "vessel_positions", + "aircraft": "aircraft_positions", +} +_CAGG = { + "vessel": "vessel_tracks_1min", + "aircraft": "aircraft_tracks_1min", +} +_ID_COL = { + "vessel": "mmsi", + "aircraft": "hex", +} + +# Last persist time per entity so AIS/ADS-B does not write every frame. +_last_write: dict[tuple[str, str], datetime] = {} +_MIN_WRITE_GAP = timedelta(seconds=20) + + +def minute_bucket(ts: datetime) -> datetime: + if ts.tzinfo is None: + ts = ts.replace(tzinfo=timezone.utc) + return ts.replace(second=0, microsecond=0) + + +def downsample_tracks(rows: list[dict]) -> list[dict]: + """Last sample per id per 1-minute bucket (mirrors the CAGG).""" + last: dict[tuple[str, datetime], dict] = {} + for row in rows: + rid = str(row.get("id") or "") + ts = row.get("ts") + if not rid or not isinstance(ts, datetime): + continue + bucket = minute_bucket(ts) + key = (rid, bucket) + prev = last.get(key) + if prev is None or ts >= prev["ts"]: + last[key] = {**row, "id": rid, "bucket": bucket, "ts": ts} + out = [] + for (_id, bucket), row in last.items(): + out.append({ + "id": row["id"], + "bucket": bucket, + "lat": row.get("lat"), + "lon": row.get("lon"), + "heading": row.get("heading"), + "speed": row.get("speed"), + "label": row.get("label"), + }) + return out + + +def positions_at_timestamp(rows: list[dict], ts: datetime) -> list[dict]: + """Positions whose 1-minute bucket equals floor(ts).""" + want = minute_bucket(ts) + picked = [r for r in downsample_tracks(rows) if r["bucket"] == want] + return [ + to_marker( + r["id"], r.get("lat"), r.get("lon"), + heading=r.get("heading"), speed=r.get("speed"), + label=r.get("label") or r["id"], + ) + for r in picked + if r.get("lat") is not None and r.get("lon") is not None + ] + + +def parse_timestamp(value: str | datetime | None) -> datetime | None: + if value is None or value == "": + return None + if isinstance(value, datetime): + ts = value + else: + raw = str(value).strip().replace("Z", "+00:00") + ts = datetime.fromisoformat(raw) + if ts.tzinfo is None: + ts = ts.replace(tzinfo=timezone.utc) + return ts + + +async def record_position(kind: Kind, marker: dict, ts: datetime | None = None) -> bool: + """Insert one sample into the raw hypertable (rate-limited).""" + vid = str(marker.get("id") or "") + lat, lon = marker.get("lat"), marker.get("lon") + if not vid or lat is None or lon is None: + return False + now = ts or datetime.now(timezone.utc) + key = (kind, vid) + prev = _last_write.get(key) + if prev is not None and now - prev < _MIN_WRITE_GAP: + return False + _last_write[key] = now + table = _RAW_TABLE[kind] + id_col = _ID_COL[kind] + extra = marker.get("extra") or {} + try: + async with async_session() as session: + await session.execute( + text( + f""" + INSERT INTO {table} ({id_col}, ts, lat, lon, heading, speed, label, extra) + VALUES (:id, :ts, :lat, :lon, :heading, :speed, :label, CAST(:extra AS jsonb)) + ON CONFLICT ({id_col}, ts) DO NOTHING + """ + ), + { + "id": vid, + "ts": now, + "lat": float(lat), + "lon": float(lon), + "heading": marker.get("heading"), + "speed": marker.get("speed"), + "label": marker.get("label") or vid, + "extra": json.dumps(extra), + }, + ) + await session.commit() + return True + except Exception: + return False + + +async def fetch_positions_at( + kind: Kind, + ts: datetime, + bbox: str | None = None, + limit: int = 2000, +) -> list[dict]: + """Read the 1-minute CAGG for the bucket containing ``ts``.""" + bucket = minute_bucket(ts) + table = _CAGG[kind] + id_col = _ID_COL[kind] + where = "bucket = :bucket" + params: dict[str, Any] = {"bucket": bucket, "limit": limit} + if bbox: + minlon, minlat, maxlon, maxlat = parse_bbox(bbox) + where += " AND lon BETWEEN :minlon AND :maxlon AND lat BETWEEN :minlat AND :maxlat" + params.update(minlon=minlon, minlat=minlat, maxlon=maxlon, maxlat=maxlat) + sql = f""" + SELECT {id_col} AS id, lat, lon, heading, speed, label, bucket + FROM {table} + WHERE {where} + LIMIT :limit + """ + try: + async with async_session() as session: + rows = (await session.execute(text(sql), params)).mappings().all() + points = [ + to_marker( + r["id"], r["lat"], r["lon"], + heading=r["heading"], speed=r["speed"], + label=r["label"] or r["id"], + extra={"bucket": r["bucket"].isoformat() if r["bucket"] else None, "dvr": True}, + ) + for r in rows + if r["lat"] is not None and r["lon"] is not None + ] + return points + except Exception: + return [] + + +async def track_range() -> dict: + """Earliest/latest buckets across both CAGGs — slider bounds.""" + try: + async with async_session() as session: + row = (await session.execute(text( + """ + SELECT + LEAST( + (SELECT min(bucket) FROM vessel_tracks_1min), + (SELECT min(bucket) FROM aircraft_tracks_1min) + ) AS tmin, + GREATEST( + (SELECT max(bucket) FROM vessel_tracks_1min), + (SELECT max(bucket) FROM aircraft_tracks_1min) + ) AS tmax + """ + ))).mappings().first() + if not row or row["tmin"] is None: + now = datetime.now(timezone.utc).replace(second=0, microsecond=0) + return {"min": (now - timedelta(hours=6)).isoformat(), "max": now.isoformat()} + return { + "min": row["tmin"].isoformat(), + "max": row["tmax"].isoformat(), + } + except Exception: + now = datetime.now(timezone.utc).replace(second=0, microsecond=0) + return {"min": (now - timedelta(hours=6)).isoformat(), "max": now.isoformat()} + + +async def recent_markers(kind: Kind, limit: int = 2000) -> list[dict]: + """Latest raw sample per id — used when in-process last-known is empty.""" + table = _RAW_TABLE[kind] + id_col = _ID_COL[kind] + sql = f""" + SELECT DISTINCT ON ({id_col}) + {id_col} AS id, lat, lon, heading, speed, label, extra + FROM {table} + WHERE ts > now() - interval '15 minutes' + ORDER BY {id_col}, ts DESC + LIMIT :limit + """ + try: + async with async_session() as session: + rows = (await session.execute(text(sql), {"limit": limit})).mappings().all() + out = [] + for r in rows: + extra = r.get("extra") or {} + if isinstance(extra, str): + try: + extra = json.loads(extra) + except (TypeError, ValueError): + extra = {} + m = to_marker( + r["id"], r["lat"], r["lon"], + heading=r["heading"], speed=r["speed"], + label=r["label"] or r["id"], + extra=extra if isinstance(extra, dict) else {}, + ) + if m.get("lat") is not None and m.get("lon") is not None: + out.append(m) + return out + except Exception: + return [] diff --git a/app/upstream_cache.py b/app/upstream_cache.py new file mode 100644 index 0000000..6f106ff --- /dev/null +++ b/app/upstream_cache.py @@ -0,0 +1,11 @@ +"""In-process TTL caches for chatty upstreams (FIRMS, RSS). No Redis.""" + +from __future__ import annotations + +from cachetools import TTLCache + +# FIRMS NRT updates every ~5–10 min; 5 min / 100 keys is enough for bbox×dataset. +firms_cache: TTLCache = TTLCache(maxsize=100, ttl=300) + +# News RSS: 1 minute is enough to absorb dashboard double-clicks / retries. +rss_cache: TTLCache = TTLCache(maxsize=100, ttl=60) diff --git a/app/ws_manager.py b/app/ws_manager.py new file mode 100644 index 0000000..b46a6b6 --- /dev/null +++ b/app/ws_manager.py @@ -0,0 +1,80 @@ +"""In-memory WebSocket pub/sub with viewport filtering. + +Zero extra deps. Ingest workers publish AIS/ADS-B points; only clients whose +current map bbox contains the point receive the payload. No Redis/Kafka. +""" + +from __future__ import annotations + +import asyncio +from typing import Any + +BBox = tuple[float, float, float, float] # minlon, minlat, maxlon, maxlat + + +def point_in_bbox(lon: float, lat: float, bbox: BBox | None) -> bool: + """True if (lon, lat) sits inside an axis-aligned viewport.""" + if bbox is None: + return False + minlon, minlat, maxlon, maxlat = bbox + return minlon <= lon <= maxlon and minlat <= lat <= maxlat + + +class ConnectionManager: + """Maps Tailscale/browser clients → viewport bbox + per-client queue.""" + + def __init__(self) -> None: + self._queues: dict[str, asyncio.Queue] = {} + self._viewports: dict[str, BBox] = {} + + def register(self, client_id: str, maxsize: int = 256) -> asyncio.Queue: + q: asyncio.Queue = asyncio.Queue(maxsize=maxsize) + self._queues[client_id] = q + return q + + def unregister(self, client_id: str) -> None: + self._queues.pop(client_id, None) + self._viewports.pop(client_id, None) + + def set_viewport(self, client_id: str, bbox: BBox) -> None: + if client_id in self._queues: + self._viewports[client_id] = bbox + + def viewport_of(self, client_id: str) -> BBox | None: + return self._viewports.get(client_id) + + def has_clients(self) -> bool: + return bool(self._queues) + + async def publish_point( + self, + kind: str, + payload: dict[str, Any], + *, + lat: float, + lon: float, + ) -> int: + """Enqueue `{type, payload}` for clients whose viewport contains the point. + + Drops the oldest queued message if a client's buffer is full so a slow + tab cannot stall ingest. Returns the number of clients that got a copy. + """ + msg = {"type": kind, "payload": payload} + sent = 0 + for client_id, queue in list(self._queues.items()): + if not point_in_bbox(lon, lat, self._viewports.get(client_id)): + continue + if queue.full(): + try: + queue.get_nowait() + except asyncio.QueueEmpty: + pass + try: + queue.put_nowait(msg) + except asyncio.QueueFull: + continue + sent += 1 + return sent + + +manager = ConnectionManager() diff --git a/docker-compose.yml b/docker-compose.yml index e03a016..b2a7cd6 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -30,7 +30,16 @@ services: POSTGRES_DB: ${DB_NAME:-osint_data} # Ensure TimescaleDB is preloaded (conf.d drop-in may be ignored by the # official image's runtime-generated postgresql.conf, so pass it explicitly). - command: ["-c", "shared_preload_libraries=timescaledb"] + # shared_buffers capped at 2GB for Pi 5 8GB / 4 cores. + command: + [ + "-c", "shared_preload_libraries=timescaledb", + "-c", "shared_buffers=2GB", + ] + deploy: + resources: + limits: + memory: 3G ports: - "127.0.0.1:5432:5432" volumes: @@ -124,6 +133,10 @@ services: AISSTREAM_IN_APP: ${AISSTREAM_IN_APP:-1} ports: - "127.0.0.1:8000:8000" + deploy: + resources: + limits: + memory: 2G healthcheck: test: ["CMD-SHELL", "python -c \"import urllib.request,sys; sys.exit(0 if urllib.request.urlopen('http://127.0.0.1:8000/api/health').status==200 else 1)\""] interval: 30s diff --git a/tests/test_bg_jobs.py b/tests/test_bg_jobs.py new file mode 100644 index 0000000..fed5541 --- /dev/null +++ b/tests/test_bg_jobs.py @@ -0,0 +1,64 @@ +"""masscan/ffmpeg stay off the request path (asyncio.create_task).""" + +from __future__ import annotations + +import asyncio + +import bg_jobs + + +def test_schedule_masscan_pass_returns_without_awaiting_scan(monkeypatch): + started = {"n": 0} + + async def slow_pass(): + started["n"] += 1 + await asyncio.sleep(30) + + monkeypatch.setattr(bg_jobs, "_run_masscan_capped", slow_pass) + bg_jobs._masscan_task = None + + async def run(): + launched = bg_jobs.schedule_masscan_pass() + assert launched is True + # Must not have blocked for the 30s pass. + assert bg_jobs._masscan_task is not None + assert not bg_jobs._masscan_task.done() + launched2 = bg_jobs.schedule_masscan_pass() + assert launched2 is False # already running + bg_jobs._masscan_task.cancel() + try: + await bg_jobs._masscan_task + except (asyncio.CancelledError, Exception): + pass + bg_jobs._masscan_task = None + + asyncio.run(run()) + + +def test_masscan_rate_cap_is_200(): + assert bg_jobs.MASSCAN_PPS_CAP == 200 + + +def test_schedule_ffmpeg_snapshot_is_a_task_not_inline(monkeypatch): + calls = {"n": 0} + + async def fake_grab(url, timeout=8.0): + calls["n"] += 1 + await asyncio.sleep(5) + return b"\xff\xd8fakejpeg" + + monkeypatch.setattr(bg_jobs, "_ffmpeg_grab", fake_grab) + bg_jobs._ffmpeg_tasks.clear() + bg_jobs._ffmpeg_cache.clear() + + async def run(): + task = bg_jobs.schedule_ffmpeg_snapshot("rtsp://10.0.0.1/") + assert isinstance(task, asyncio.Task) + assert not task.done() + task.cancel() + try: + await task + except (asyncio.CancelledError, Exception): + pass + + asyncio.run(run()) diff --git a/tests/test_fire_aircraft.py b/tests/test_fire_aircraft.py new file mode 100644 index 0000000..692ee7b --- /dev/null +++ b/tests/test_fire_aircraft.py @@ -0,0 +1,66 @@ +"""WFIGS/FIRMS × firefighting ADS-B correlation within 20 miles.""" + +from __future__ import annotations + +from fire_aircraft import ( + FIREFIGHTER_ICAO, + RADIUS_MILES, + correlate_aircraft_to_fires, + is_firefighter, +) + + +def _ac(hex_id, lat, lon, icao, **extra): + return { + "id": hex_id, + "lat": lat, + "lon": lon, + "label": hex_id, + "heading": 0, + "speed": 120, + "extra": {"type": icao, "hex": hex_id, **extra}, + } + + +def _fire(name, lat, lon, **extra): + return { + "id": name, + "lat": lat, + "lon": lon, + "label": name, + "extra": extra, + } + + +def test_air_tractor_is_firefighter_airliner_is_not(): + assert is_firefighter(_ac("aaa", 0, 0, "AT802")) is True + assert is_firefighter(_ac("bbb", 0, 0, "C130")) is True + assert is_firefighter(_ac("ccc", 0, 0, "B738")) is False + assert "AT802" in FIREFIGHTER_ICAO + + +def test_flags_tanker_within_20_miles_of_fire(): + # ~10 miles north of a Piedmont fire + fire = _fire("Jones Gap", 35.00, -82.00) + tanker = _ac("acf001", 35.145, -82.00, "AT802") + airliner = _ac("a0b738", 35.145, -82.00, "B738") + far = _ac("acfar", 35.50, -82.00, "C130") # ~34 miles + hits = correlate_aircraft_to_fires([fire], [tanker, airliner, far]) + assert RADIUS_MILES == 20.0 + assert len(hits) == 1 + h = hits[0] + assert h["aircraft_hex"] == "acf001" + assert h["fire_id"] == "Jones Gap" + assert h["aircraft_type"] == "AT802" + assert 0 < h["distance_mi"] <= 20.0 + + +def test_persist_shape_has_reload_fields(): + fire = _fire("Jones Gap", 35.00, -82.00, src="wfigs") + tanker = _ac("acf001", 35.10, -82.00, "S64") + hits = correlate_aircraft_to_fires([fire], [tanker]) + assert set(hits[0]).issuperset({ + "fire_id", "fire_lat", "fire_lon", + "aircraft_hex", "aircraft_type", "aircraft_lat", "aircraft_lon", + "distance_mi", + }) diff --git a/tests/test_fire_ingest.py b/tests/test_fire_ingest.py index 31f05d2..b6e7dc1 100644 --- a/tests/test_fire_ingest.py +++ b/tests/test_fire_ingest.py @@ -85,3 +85,87 @@ def test_ingest_fire_row_drops_malformed(clean_fires): assert await ingest_fire_row(make_fire_msg(acq_time="garbage")) is False asyncio.run(run()) + + +class _InsertSession: + """async_session stand-in: fire insert succeeds (rowcount=1).""" + + def __init__(self): + self.rowcount = 1 + + async def execute(self, *a, **k): + return self + + async def commit(self): + return None + + async def __aenter__(self): + return self + + async def __aexit__(self, *a): + return False + + +def test_ingest_fire_row_geofence_when_cache_empty(monkeypatch): + """Ingester process has empty geofence cache; still notify on FIRMS insert.""" + import geofence + import ingestor + import live_layers + + geofence._cache.clear() + live_layers.aircraft_last_known.clear() + notified = [] + + async def fake_record(**kw): + notified.append(kw) + return 1 + + async def no_markers(*a, **k): + return [] + + monkeypatch.setattr(ingestor, "async_session", _InsertSession) + monkeypatch.setattr(geofence, "record_and_notify", fake_record) + monkeypatch.setattr("tracks.recent_markers", no_markers) + + assert asyncio.run(ingest_fire_row(make_fire_msg())) is True + assert len(notified) == 1 + assert notified[0]["source_kind"] == "firms" + assert notified[0]["lat"] == 39.45678 + assert notified[0]["lon"] == -121.12345 + + +def test_ingest_fire_row_correlates_from_hypertable_when_last_known_empty(monkeypatch): + """FIRMS ingester has no ADS-B last-known; still correlate from aircraft_positions.""" + import geofence + import ingestor + import live_layers + + geofence._cache.clear() + live_layers.aircraft_last_known.clear() + correlated = [] + + async def fake_record(**kw): + return 0 + + async def fake_recent(kind, limit=2000): + assert kind == "aircraft" + return [{ + "id": "acf001", + "lat": 39.45, + "lon": -121.12, + "extra": {"type": "AT802"}, + }] + + async def fake_corr(fires, aircraft): + correlated.append((fires, aircraft)) + return aircraft + + monkeypatch.setattr(ingestor, "async_session", _InsertSession) + monkeypatch.setattr(geofence, "record_and_notify", fake_record) + monkeypatch.setattr("tracks.recent_markers", fake_recent) + monkeypatch.setattr("fire_aircraft.correlate_and_notify", fake_corr) + + assert asyncio.run(ingest_fire_row(make_fire_msg())) is True + assert len(correlated) == 1 + assert correlated[0][1][0]["id"] == "acf001" + assert correlated[0][0][0]["lat"] == 39.45678 diff --git a/tests/test_fire_sources.py b/tests/test_fire_sources.py index 3a3e4af..c787ec2 100644 --- a/tests/test_fire_sources.py +++ b/tests/test_fire_sources.py @@ -83,6 +83,8 @@ def test_ingest_fires_idles_without_map_key(monkeypatch): def test_ingest_fires_uses_keystore_key(monkeypatch): # Key saved via the dashboard Keys page (Postgres) is picked up. + from upstream_cache import firms_cache + firms_cache.clear() monkeypatch.setenv("FIRMS_MAP_KEY", "") monkeypatch.setattr( "fire_sources.get_api_key", diff --git a/tests/test_geofence.py b/tests/test_geofence.py new file mode 100644 index 0000000..df1d59d --- /dev/null +++ b/tests/test_geofence.py @@ -0,0 +1,136 @@ +"""Geofence hit detection and WS alert routing (no Redis).""" + +from __future__ import annotations + +import asyncio + +import geofence +from geofence import ( + matching_geofences, + point_in_geojson, + validate_polygon_geojson, +) +from ws_manager import ConnectionManager + + +NC_BOX = { + "type": "Polygon", + "coordinates": [[ + [-80.0, 35.0], + [-78.0, 35.0], + [-78.0, 36.0], + [-80.0, 36.0], + [-80.0, 35.0], + ]], +} + + +def test_point_inside_polygon_hits(): + assert point_in_geojson(-79.0, 35.5, NC_BOX) is True + + +def test_point_outside_polygon_misses(): + assert point_in_geojson(-122.4, 37.7, NC_BOX) is False + + +def test_validate_rejects_non_polygon(): + try: + validate_polygon_geojson({"type": "Point", "coordinates": [-79.0, 35.5]}) + assert False, "expected ValueError" + except ValueError: + pass + + +def test_matching_geofences_only_active_hits(): + fences = [ + {"id": "a", "name": "NC", "active": True, "geojson": NC_BOX}, + {"id": "b", "name": "off", "active": False, "geojson": NC_BOX}, + ] + hits = matching_geofences(-79.0, 35.5, fences) + assert [h["id"] for h in hits] == ["a"] + assert matching_geofences(-122.4, 37.7, fences) == [] + + +def test_geofence_alert_fans_out_only_to_viewport_clients(): + mgr = ConnectionManager() + q_nc = mgr.register("nc") + q_sf = mgr.register("sf") + mgr.set_viewport("nc", (-80.0, 35.0, -78.0, 36.0)) + mgr.set_viewport("sf", (-123.0, 37.0, -121.0, 38.0)) + + async def run(): + payload = { + "geofence_id": "a", + "geofence_name": "NC", + "source_kind": "ais", + "entity_id": "366123456", + "lat": 35.5, + "lon": -79.0, + } + n = await mgr.publish_point("geofence_alert", payload, lat=35.5, lon=-79.0) + assert n == 1 + msg = q_nc.get_nowait() + assert msg["type"] == "geofence_alert" + assert msg["payload"]["entity_id"] == "366123456" + assert q_sf.empty() + + asyncio.run(run()) + + +def test_record_and_notify_queries_postgis_when_cache_empty(monkeypatch): + """FIRMS ingest in the ingester has an empty in-process cache — still ST_Intersects.""" + import geofence + + geofence._cache.clear() + geofence._recent_hits.clear() + st_called = [] + + async def fake_st(lon, lat): + st_called.append((lon, lat)) + return [{ + "id": "11111111-1111-1111-1111-111111111111", + "name": "NC", + "geojson": NC_BOX, + "active": True, + }] + + monkeypatch.setattr(geofence, "st_intersects", fake_st) + + executed: list = [] + + class FakeSession: + async def execute(self, stmt, params=None): + executed.append(params or {}) + return None + + async def commit(self): + executed.append("commit") + + async def __aenter__(self): + return self + + async def __aexit__(self, *a): + return False + + monkeypatch.setattr(geofence, "async_session", FakeSession) + + async def run(): + from ws_manager import manager + q = manager.register("nc") + manager.set_viewport("nc", (-80.0, 35.0, -78.0, 36.0)) + n = await geofence.record_and_notify( + source_kind="firms", entity_id="35.5,-79.0,N", + lat=35.5, lon=-79.0, payload={"satellite": "N"}, + ) + msg = None if q.empty() else q.get_nowait() + manager.unregister("nc") + return n, msg + + n, msg = asyncio.run(run()) + assert st_called == [(-79.0, 35.5)] + assert n == 1 + assert msg["type"] == "geofence_alert" + assert msg["payload"]["source_kind"] == "firms" + inserts = [p for p in executed if isinstance(p, dict)] + assert inserts and inserts[0]["source_kind"] == "firms" + assert "commit" in executed diff --git a/tests/test_phase1_guardrails.py b/tests/test_phase1_guardrails.py new file mode 100644 index 0000000..de2a00c --- /dev/null +++ b/tests/test_phase1_guardrails.py @@ -0,0 +1,33 @@ +"""Phase 1 guardrails: compose limits, 500ms debounce, no extra brokers.""" + +from __future__ import annotations + +from pathlib import Path + +ROOT = Path(__file__).resolve().parent.parent + + +def test_compose_memory_limits_and_shared_buffers(): + text = (ROOT / "docker-compose.yml").read_text() + assert "shared_buffers=2GB" in text + assert "shared_preload_libraries=timescaledb" in text + assert "memory: 3G" in text or "memory: 3GB" in text + assert "memory: 2G" in text or "memory: 2GB" in text + + +def test_map_moveend_debounced_500ms(): + html = (ROOT / "app" / "static" / "index.html").read_text() + assert "map.on('moveend'" in html + # Existing 300ms debounce must be 500ms so pans don't spam bbox POSTs/WS. + assert "}, 500);" in html + assert "}, 300);" not in html.split("map.on('moveend'")[1][:800] + + +def test_no_redis_kafka_celery(): + req = (ROOT / "app" / "requirements.txt").read_text().lower() + compose = (ROOT / "docker-compose.yml").read_text().lower() + for blob in (req, compose): + assert "redis" not in blob + assert "kafka" not in blob + assert "celery" not in blob + assert "cachetools" in req diff --git a/tests/test_phase2_api.py b/tests/test_phase2_api.py new file mode 100644 index 0000000..e52548b --- /dev/null +++ b/tests/test_phase2_api.py @@ -0,0 +1,51 @@ +"""Phase 2 API contracts (no Redis; geofences ≠ /api/alerts).""" + +from __future__ import annotations + +import asyncio + +import httpx + +from main import app + +BASE = "http://test" + + +async def _req(method: str, path: str, **kw) -> httpx.Response: + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url=BASE) as client: + return await client.request(method, path, **kw) + + +def test_geofence_post_rejects_point(): + resp = asyncio.run(_req( + "POST", "/api/geofences", + json={"name": "nope", "geojson": {"type": "Point", "coordinates": [-79, 35]}}, + )) + assert resp.status_code == 422 + + +def test_geofences_list_does_not_collide_with_alerts(): + resp = asyncio.run(_req("GET", "/api/geofences")) + assert resp.status_code == 200 + assert isinstance(resp.json(), list) + + +def test_aircraft_timestamp_returns_list_not_live_error(): + resp = asyncio.run(_req( + "GET", "/api/aircraft", + params={"bbox": "-80,35,-78,36", "timestamp": "2026-08-28T12:04:00Z"}, + )) + assert resp.status_code == 200 + assert isinstance(resp.json(), list) + + +def test_tracks_range_has_bounds(): + body = asyncio.run(_req("GET", "/api/tracks/range")).json() + assert "min" in body and "max" in body + + +def test_fire_aircraft_reload_endpoint(): + resp = asyncio.run(_req("GET", "/api/fire-aircraft")) + assert resp.status_code == 200 + assert isinstance(resp.json(), list) diff --git a/tests/test_tracks.py b/tests/test_tracks.py new file mode 100644 index 0000000..4ef6a5d --- /dev/null +++ b/tests/test_tracks.py @@ -0,0 +1,50 @@ +"""1-minute downsampled tracks for DVR playback.""" + +from __future__ import annotations + +from datetime import datetime, timezone + +from tracks import ( + TRACK_BUCKET, + downsample_tracks, + minute_bucket, + positions_at_timestamp, +) + + +def _ts(h, m, s=0): + return datetime(2026, 8, 28, h, m, s, tzinfo=timezone.utc) + + +def test_minute_bucket_floors_seconds(): + assert minute_bucket(_ts(12, 4, 47)) == _ts(12, 4, 0) + assert TRACK_BUCKET == "1 minute" + + +def test_downsample_keeps_last_sample_per_id_per_minute(): + rows = [ + {"id": "a1", "ts": _ts(12, 4, 10), "lat": 35.0, "lon": -79.0}, + {"id": "a1", "ts": _ts(12, 4, 50), "lat": 35.1, "lon": -79.1}, + {"id": "a1", "ts": _ts(12, 5, 5), "lat": 35.2, "lon": -79.2}, + {"id": "b2", "ts": _ts(12, 4, 20), "lat": 36.0, "lon": -80.0}, + ] + out = downsample_tracks(rows) + by = {(r["id"], r["bucket"]): r for r in out} + assert by[("a1", _ts(12, 4))]["lat"] == 35.1 + assert by[("a1", _ts(12, 5))]["lat"] == 35.2 + assert by[("b2", _ts(12, 4))]["lat"] == 36.0 + assert len(out) == 3 + + +def test_positions_at_timestamp_uses_that_minute_window(): + rows = [ + {"id": "a1", "ts": _ts(12, 4, 50), "lat": 35.1, "lon": -79.1, "heading": 90, "speed": 10, "label": "A1"}, + {"id": "a1", "ts": _ts(12, 5, 5), "lat": 35.2, "lon": -79.2, "heading": 91, "speed": 11, "label": "A1"}, + {"id": "b2", "ts": _ts(12, 4, 20), "lat": 36.0, "lon": -80.0, "heading": 0, "speed": 0, "label": "B2"}, + ] + at = positions_at_timestamp(rows, _ts(12, 4, 59)) + ids = {p["id"]: p for p in at} + assert ids["a1"]["lat"] == 35.1 + assert ids["b2"]["lat"] == 36.0 + later = positions_at_timestamp(rows, _ts(12, 5, 30)) + assert {p["id"]: p["lat"] for p in later} == {"a1": 35.2} diff --git a/tests/test_upstream_cache.py b/tests/test_upstream_cache.py new file mode 100644 index 0000000..040527b --- /dev/null +++ b/tests/test_upstream_cache.py @@ -0,0 +1,104 @@ +"""FIRMS / RSS in-process TTLCache — no Redis.""" + +from __future__ import annotations + +import asyncio + +from cachetools import TTLCache + +from fire_sources import ingest_fires +from sources import ingest_rss_feed +from test_fire_sources import SAMPLE_CSV, _async_return +from upstream_cache import firms_cache, rss_cache + + +def test_firms_and_rss_caches_are_ttlcache(): + assert isinstance(firms_cache, TTLCache) + assert firms_cache.maxsize == 100 + assert firms_cache.ttl == 300 + assert isinstance(rss_cache, TTLCache) + assert rss_cache.maxsize == 100 + assert 60 <= rss_cache.ttl <= 300 + + +def test_ingest_fires_hits_http_once_within_ttl(monkeypatch): + firms_cache.clear() + monkeypatch.setenv("FIRMS_MAP_KEY", "k" * 32) + monkeypatch.setenv("FIRMS_DATASETS", "VIIRS_NOAA20_NRT") + # fire_sources already imported FIRMS_DATASETS — patch the module attr + monkeypatch.setattr("fire_sources.FIRMS_DATASETS", ["VIIRS_NOAA20_NRT"]) + + hits = {"n": 0} + + class FakeResp: + text = SAMPLE_CSV + + def raise_for_status(self): + pass + + class FakeClient: + def __init__(self, **kw): + pass + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + async def get(self, url): + hits["n"] += 1 + return FakeResp() + + monkeypatch.setattr("fire_sources.httpx.AsyncClient", FakeClient) + + async def fake_publish(points): + return len(points) + + monkeypatch.setattr("fire_sources.publish_fire_batch", fake_publish) + + assert asyncio.run(ingest_fires()) == 5 + assert asyncio.run(ingest_fires()) == 5 + assert hits["n"] == 1 + + +def test_ingest_rss_hits_http_once_within_ttl(monkeypatch): + rss_cache.clear() + hits = {"n": 0} + + class FakeResp: + text = """ + t + hellohttp://x.example/1 + body""" + + def raise_for_status(self): + pass + + class FakeClient: + def __init__(self, **kw): + pass + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + async def get(self, url): + hits["n"] += 1 + return FakeResp() + + monkeypatch.setattr("sources.httpx.AsyncClient", FakeClient) + + published = [] + + async def fake_publish(subject, event): + published.append((subject, event)) + + monkeypatch.setattr("sources.publish_event", fake_publish) + + assert asyncio.run(ingest_rss_feed("http://feeds.example/rss")) == 1 + assert asyncio.run(ingest_rss_feed("http://feeds.example/rss")) == 1 + assert hits["n"] == 1 + assert published[0][0] == "events.rss" diff --git a/tests/test_ws_manager.py b/tests/test_ws_manager.py new file mode 100644 index 0000000..23be057 --- /dev/null +++ b/tests/test_ws_manager.py @@ -0,0 +1,91 @@ +"""Viewport-filtered in-memory pub/sub — no Redis.""" + +from __future__ import annotations + +import asyncio + +from ws_manager import ConnectionManager, point_in_bbox + + +def test_point_in_bbox_inclusive(): + box = (-80.0, 35.0, -78.0, 36.0) + assert point_in_bbox(-79.0, 35.5, box) is True + assert point_in_bbox(-80.0, 35.0, box) is True + assert point_in_bbox(-77.0, 35.5, box) is False + assert point_in_bbox(-79.0, 34.0, box) is False + + +def test_missing_viewport_does_not_get_firehose(): + mgr = ConnectionManager() + q = mgr.register("tailnet-a") + + async def run(): + n = await mgr.publish_point("ais", {"id": "mmsi-1"}, lat=35.5, lon=-79.0) + assert n == 0 + assert q.empty() + + asyncio.run(run()) + + +def test_fanout_only_to_intersecting_viewports(): + mgr = ConnectionManager() + q_nc = mgr.register("nc") + q_sf = mgr.register("sf") + mgr.set_viewport("nc", (-80.0, 35.0, -78.0, 36.0)) + mgr.set_viewport("sf", (-123.0, 37.0, -121.0, 38.0)) + + async def run(): + n = await mgr.publish_point( + "ais", {"id": "mmsi-1", "lat": 35.5, "lon": -79.0}, + lat=35.5, lon=-79.0, + ) + assert n == 1 + msg = q_nc.get_nowait() + assert msg["type"] == "ais" + assert msg["payload"]["id"] == "mmsi-1" + assert q_sf.empty() + + n2 = await mgr.publish_point( + "adsb", {"id": "a1b2c3"}, lat=37.7, lon=-122.4, + ) + assert n2 == 1 + msg2 = q_sf.get_nowait() + assert msg2["type"] == "adsb" + assert q_nc.empty() + + asyncio.run(run()) + + +def test_disconnect_stops_fanout(): + mgr = ConnectionManager() + q = mgr.register("gone") + mgr.set_viewport("gone", (-180.0, -90.0, 180.0, 90.0)) + mgr.unregister("gone") + + async def run(): + n = await mgr.publish_point("ais", {"id": "x"}, lat=0.0, lon=0.0) + assert n == 0 + assert q.empty() + + asyncio.run(run()) + + +def test_upsert_vessel_fans_out_to_intersecting_client(monkeypatch): + from live_layers import upsert_vessel, vessel_last_known + from ws_manager import manager + + vessel_last_known.clear() + manager._queues.clear() + manager._viewports.clear() + q = manager.register("nc") + manager.set_viewport("nc", (-80.0, 35.0, -78.0, 36.0)) + + async def run(): + await upsert_vessel({"id": "366123456", "lat": 35.2, "lon": -79.1, "label": "TEST"}) + msg = q.get_nowait() + assert msg["type"] == "ais" + assert msg["payload"]["id"] == "366123456" + + asyncio.run(run()) + manager.unregister("nc") + vessel_last_known.clear() -- 2.45.3