From c48788d4b6140ed2479254305bf3ef8ccb012d94 Mon Sep 17 00:00:00 2001 From: Sirius DevOps Date: Sat, 29 Aug 2026 20:40:27 -0400 Subject: [PATCH] fix(map): geofence delete, vessel snapshots, sentinel cache, news briefs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Geofences could be drawn but not removed. VesselAPI Hormuz dots vanished on restart and DVR skipped between the 5 daily polls. Sentinel-1 re-hit STAC on every pan and often painted a neighbouring swath. Executive briefs truncated; ticker stayed empty unless something was critical. - Layer-panel list + polygon popup DELETE /api/geofences/{id} - Persist VesselAPI polls to vessels (UTC-day purge, DVR as-of, boot hydrate) - Cache Sentinel-1 by 2° cell; pick covering scene; clip Leaflet tiles - Retry truncated LLM JSON; ticker falls back to medium/low; 3-min HUD poll --- alembic/versions/009_vessels.py | 41 ++++ app/live_layers.py | 63 +++++- app/main.py | 33 +++- app/static/index.html | 82 +++++++- app/tracks.py | 17 +- app/vesselapi.py | 228 +++++++++++++++++++++- docs/news.md | 17 +- news/summerizer/intel.py | 21 +- news/summerizer/nous_client.py | 37 +++- news/summerizer/summarizer.py | 24 ++- news/summerizer/tests/test_intel.py | 10 + news/summerizer/tests/test_nous_client.py | 27 +++ news/summerizer/tests/test_prompts.py | 8 + tests/test_api_news.py | 15 ++ tests/test_frontend_reliability.py | 8 + tests/test_geofence_frontend.py | 26 +++ tests/test_live_layers.py | 23 ++- tests/test_sentinel1_frontend.py | 8 + tests/test_vesselapi.py | 2 + tests/test_vessels_snapshot.py | 36 ++++ 20 files changed, 669 insertions(+), 57 deletions(-) create mode 100644 alembic/versions/009_vessels.py create mode 100644 tests/test_geofence_frontend.py create mode 100644 tests/test_vessels_snapshot.py diff --git a/alembic/versions/009_vessels.py b/alembic/versions/009_vessels.py new file mode 100644 index 0000000..8f0dd01 --- /dev/null +++ b/alembic/versions/009_vessels.py @@ -0,0 +1,41 @@ +"""vessels — daily VesselAPI snapshots for DVR as-of + +Revision ID: 009_vessels +Revises: 008_summary_kind +Create Date: 2026-08-29 +""" + +from alembic import op + +revision = "009_vessels" +down_revision = "008_summary_kind" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.execute( + """ + CREATE TABLE IF NOT EXISTS vessels ( + mmsi TEXT NOT NULL, + poll_at 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, poll_at) + ) + """ + ) + op.execute( + "CREATE INDEX IF NOT EXISTS ix_vessels_poll_at ON vessels (poll_at DESC)" + ) + op.execute( + "CREATE INDEX IF NOT EXISTS ix_vessels_bbox ON vessels (lon, lat)" + ) + + +def downgrade() -> None: + op.execute("DROP TABLE IF EXISTS vessels") diff --git a/app/live_layers.py b/app/live_layers.py index 2731047..fa632b1 100644 --- a/app/live_layers.py +++ b/app/live_layers.py @@ -51,7 +51,9 @@ PC_SAS_TOKEN = "https://planetarycomputer.microsoft.com/api/sas/v1/token/sentine # nginx vhost. Relative template — Leaflet resolves it against the page origin, # so the browser never touches a raw loopback port or titiler.xyz. TITILER_COG_TILES = f"{TITILER_PUBLIC_BASE}/cog/tiles/WebMercatorQuad/{{z}}/{{x}}/{{y}}" -SENTINEL1_TTL = 20 * 60 # 15–30 min quota-friendly window +SENTINEL1_TTL = 6 * 3600 # S-1 revisit is 6–12 days; cache the COG all afternoon +SENTINEL1_CELL = 2.0 # degrees — pan/zoom inside a cell reuses the same scene +SENTINEL1_STAC_LIMIT = 8 SENTINEL1_ATTRIBUTION = "Copernicus Sentinel-1 / Microsoft Planetary Computer" # GPSJAM (John Wiseman / ADS-B Exchange): daily H3 hexes of aircraft nav @@ -1244,25 +1246,71 @@ def sentinel1_tile_url(signed_cog: str) -> str: return f"{TITILER_COG_TILES}?{params}" +def feature_bbox(feat: dict) -> list[float] | None: + """STAC Feature bbox as [minlon, minlat, maxlon, maxlat], or None.""" + raw = feat.get("bbox") if isinstance(feat, dict) else None + if isinstance(raw, (list, tuple)) and len(raw) >= 4: + try: + return [float(raw[0]), float(raw[1]), float(raw[2]), float(raw[3])] + except (TypeError, ValueError): + pass + geom = (feat or {}).get("geometry") or {} + coords = geom.get("coordinates") if isinstance(geom, dict) else None + if not coords: + return None + lons: list[float] = [] + lats: list[float] = [] + + def _walk(node: Any) -> None: + if isinstance(node, (list, tuple)) and node and isinstance(node[0], (int, float)): + lons.append(float(node[0])) + lats.append(float(node[1])) + elif isinstance(node, (list, tuple)): + for child in node: + _walk(child) + + _walk(coords) + if not lons: + return None + return [min(lons), min(lats), max(lons), max(lats)] + + +def pick_sentinel_feature(features: list, lon: float, lat: float) -> dict | None: + """Prefer the scene whose bbox covers the viewport center; else first.""" + if not features: + return None + for feat in features: + bb = feature_bbox(feat) + if bb and bb[0] <= lon <= bb[2] and bb[1] <= lat <= bb[3]: + return feat + return features[0] + + async def fetch_sentinel1(bbox: str) -> dict | None: """Most recent Sentinel-1 GRD COG for a viewport, signed and TiTiler-ready. Returns the overlay tile-template dict, or ``None`` when no GRD imagery covers the bbox in the last 7 days (caller maps to 404). Queries Planetary - Computer only when called; cached per quantized bbox + UTC day. + Computer only when called; cached per 2° cell + UTC day so pan/zoom inside + the same region reuses the COG instead of picking a neighbouring swath. """ minlon, minlat, maxlon, maxlat = parse_bbox(bbox) + qminlon, qminlat, qmaxlon, qmaxlat = quantize_bbox( + minlon, minlat, maxlon, maxlat, step=SENTINEL1_CELL, + ) day = datetime.now(timezone.utc).date().isoformat() - key = f"sentinel1:{day}:{bbox_cell_key(bbox)}" + key = f"sentinel1:{day}:{qminlon:.4f},{qminlat:.4f},{qmaxlon:.4f},{qmaxlat:.4f}" + clon = (qminlon + qmaxlon) / 2.0 + clat = (qminlat + qmaxlat) / 2.0 async def _load() -> dict | None: now = datetime.now(timezone.utc) week_ago = now - timedelta(days=7) payload = { "collections": ["sentinel-1-grd"], - "bbox": [minlon, minlat, maxlon, maxlat], + "bbox": [qminlon, qminlat, qmaxlon, qmaxlat], "datetime": f"{week_ago.isoformat()}/{now.isoformat()}", - "limit": 1, + "limit": SENTINEL1_STAC_LIMIT, "sortby": [{"field": "datetime", "direction": "desc"}], } data = await _pc_call(_post_json(PC_STAC_SEARCH, json=payload)) @@ -1270,7 +1318,9 @@ async def fetch_sentinel1(bbox: str) -> dict | None: if not features: return None - feat = features[0] + feat = pick_sentinel_feature(features, clon, clat) + if not feat: + return None assets = feat.get("assets") or {} chosen_href: str | None = None polarization: str | None = None @@ -1299,6 +1349,7 @@ async def fetch_sentinel1(bbox: str) -> dict | None: "datetime": props.get("datetime") or feat.get("datetime"), "polarization": polarization, "attribution": SENTINEL1_ATTRIBUTION, + "bbox": feature_bbox(feat), } return await _ttl_get(key, float(SENTINEL1_TTL), _load) diff --git a/app/main.py b/app/main.py index dd799d7..2d12e25 100644 --- a/app/main.py +++ b/app/main.py @@ -74,6 +74,11 @@ async def _lifespan(app: FastAPI): await refresh_cache() except Exception: pass + try: + from vesselapi import hydrate_last_known + await hydrate_last_known() + except Exception: + pass from config import AISSTREAM_IN_APP, VESSELAPI_IN_APP ais_task = None vesselapi_task = None @@ -1263,6 +1268,7 @@ async def list_news_summaries( _FLAGGED = ("critical", "high") +_LESSER = ("medium", "low") @app.get("/api/news/ticker", response_model=list[NewsTickerItemOut]) @@ -1273,7 +1279,7 @@ async def list_news_ticker( ), limit: int = Query(20, ge=1, le=50), ): - """Flagged ticker rows (critical/high), newest first. No LLM required.""" + """Ticker rows: critical/high first; medium/low if nothing is flagged.""" async with async_session() as session: stmt = ( select(news_items) @@ -1287,6 +1293,19 @@ async def list_news_ticker( stmt = stmt.where(news_items.c.created_at >= since) stmt = stmt.limit(limit) rows = (await session.execute(stmt)).mappings().all() + if not rows: + stmt = ( + select(news_items) + .where( + news_items.c.kind == "ticker", + news_items.c.importance.in_(_LESSER), + ) + .order_by(news_items.c.created_at.desc()) + ) + if since: + stmt = stmt.where(news_items.c.created_at >= since) + stmt = stmt.limit(limit) + rows = (await session.execute(stmt)).mappings().all() return [ NewsTickerItemOut( id=r["id"], headline=r["headline"], importance=r["importance"], @@ -1530,7 +1549,17 @@ async def list_vessels( 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) + from vesselapi import fetch_vessels_as_of + if src == "vesselapi": + return overlay_json(await fetch_vessels_as_of(ts, bbox, limit), 5) + ais = await fetch_positions_at("vessel", ts, bbox, limit) + if src == "aisstream": + return overlay_json(ais, 5) + va = await fetch_vessels_as_of(ts, bbox, limit) + by_id = {m["id"]: m for m in ais} + for m in va: + by_id[m["id"]] = m + return overlay_json(list(by_id.values())[:limit], 5) return overlay_json(await fetch_vessels(bbox, limit, src=src), 5) except ValueError as exc: raise HTTPException(422, str(exc)) from exc diff --git a/app/static/index.html b/app/static/index.html index 4100be3..02122d1 100644 --- a/app/static/index.html +++ b/app/static/index.html @@ -358,6 +358,11 @@ .lp-future .lp-name { cursor: not-allowed; } .lp-error { color: var(--red); font-size: 0.66rem; } .lp-note { font-size: 0.6rem; color: var(--muted); opacity: 0.85; line-height: 1.35; } + .gf-list { display: flex; flex-direction: column; gap: 0.22rem; max-height: 8rem; overflow-y: auto; } + .gf-item { display: flex; justify-content: space-between; align-items: center; gap: 0.4rem; font-size: 0.64rem; } + .gf-item span { overflow: hidden; text-overflow: ellipsis; white-space: nowrap; color: var(--text); } + .gf-del { background: transparent; border: 1px solid var(--magenta); color: var(--magenta); font-family: 'Share Tech Mono', monospace; font-size: 0.58rem; letter-spacing: 0.08em; text-transform: uppercase; padding: 0.12rem 0.35rem; border-radius: 3px; cursor: pointer; } + .gf-del:hover { background: rgba(255,46,151,0.16); } /* ── Camera / blip popup thumbnails ── */ .cam-pop { min-width: 210px; max-width: 260px; } @@ -583,6 +588,8 @@ .tick-item .b-tag { font-family: 'Share Tech Mono', monospace; font-size: 0.6rem; color: #1c0311; background: var(--magenta); border-radius: 2px; padding: 0.08rem 0.4rem; letter-spacing: 0.1em; } .tick-item .b-tag.critical { background: var(--red); } .tick-item .b-tag.high { background: var(--amber); } + .tick-item .b-tag.medium { background: var(--cyan); color: #041018; } + .tick-item .b-tag.low { background: var(--muted); color: #041018; } .tick-item.brief:hover { color: var(--magenta); } .tick-item.standby { color: var(--muted); opacity: 0.75; } .tick-item.standby .price { color: var(--muted); } @@ -906,7 +913,8 @@
-
No Leaflet.Draw — click vertices, double-click to close. Saved to POST /api/geofences.
+
+
Click vertices, double-click to close. Delete a saved fence from the list or its popup.
@@ -1420,7 +1428,7 @@ function initMarketTicker() { /* ═══════════════ NEWS FEED + TICKER ═══════════════ */ let newsInterval = null; -const NEWS_REFRESH_MS = 900000; // 15-min cycle +const NEWS_REFRESH_MS = 180000; // 3-min ticker/brief poll let newsTickerBuilt = false; function newsEsc(s) { @@ -1527,10 +1535,11 @@ function renderNewsTicker(articles, summaries, tickerItems) { if (ticks.length) { ticks.forEach(t => { const imp = String(t.importance || '').toLowerCase(); - const tag = imp === 'critical' ? 'CRITICAL' : 'HIGH'; + const tag = imp === 'critical' ? 'CRITICAL' : imp === 'high' ? 'HIGH' : imp === 'medium' ? 'MED' : 'LOW'; + const tagClass = (imp === 'critical' || imp === 'high' || imp === 'medium' || imp === 'low') ? imp : 'high'; const loc = (t.location_name || '').trim(); const inner = - `${tag}` + + `${tag}` + `${newsEsc((t.headline || '').trim())}` + (loc ? `${newsEsc(loc)}` : '') + `${newsTimeAgo(t.created_at)}▸`; @@ -1906,6 +1915,7 @@ let extraAttribs = new Set(); let radarLayer = null, radarOn = true, radarOpacity = 0.7, radarMeta = null, radarTimer = null; let thermalLayer = null, thermalOn = false; let sentinelLayer = null, sentinelOn = false, sentinelOpacity = 0.8; +let sentinelItemId = null, sentinelBounds = null; let wxAlertsGroup = null, wxAlertsOn = true; let perimGroup = null, perimOn = true; let incidentsGroup = null, incidentsOn = false; @@ -3071,6 +3081,8 @@ async function toggleSentinel1() { if (sentinelOn) await loadSentinel1(); else { sentinelLayer = dropLayer(sentinelLayer); + sentinelItemId = null; + sentinelBounds = null; const n = document.getElementById('lp-sentinel-count'); if (n) n.textContent = '7d'; } @@ -3080,8 +3092,15 @@ function setSentinelOpacity(v) { document.getElementById('lp-sentinel-val').textContent = `${Math.round(v)}%`; if (sentinelLayer) sentinelLayer.setOpacity(sentinelOpacity); } +function sentinelStillCovers() { + if (!map || !sentinelBounds || sentinelBounds.length < 4) return false; + const c = map.getCenter(); + return c.lng >= sentinelBounds[0] && c.lng <= sentinelBounds[2] + && c.lat >= sentinelBounds[1] && c.lat <= sentinelBounds[3]; +} async function loadSentinel1() { if (!map || !sentinelOn) return; + if (sentinelLayer && sentinelStillCovers()) return; const req = ++overlayReq.sar; const countEl = document.getElementById('lp-sentinel-count'); const hint = document.getElementById('map-hint'); @@ -3093,6 +3112,8 @@ async function loadSentinel1() { if (req !== overlayReq.sar) return; if (r.status === 404 && body && body.error === 'no_imagery') { sentinelLayer = dropLayer(sentinelLayer); + sentinelItemId = null; + sentinelBounds = null; if (hint) hint.textContent = 'No Sentinel-1 imagery for this view in the last 7 days.'; if (countEl) countEl.textContent = 'none'; return; @@ -3109,13 +3130,31 @@ async function loadSentinel1() { if (countEl) countEl.textContent = 'err'; return; } + if (body.itemId && body.itemId === sentinelItemId && sentinelLayer) { + if (Array.isArray(body.bbox) && body.bbox.length >= 4) sentinelBounds = body.bbox; + sentinelLayer.setOpacity(sentinelOpacity); + if (countEl) countEl.textContent = String(body.polarization || 'SAR').toUpperCase(); + return; + } const attrib = body.attribution || ''; - sentinelLayer = dropLayer(sentinelLayer); - sentinelLayer = L.tileLayer(body.tileUrl, { + const layerOpts = { opacity: sentinelOpacity, maxZoom: 18, attribution: attrib, - }).addTo(map); + noWrap: true, + }; + if (Array.isArray(body.bbox) && body.bbox.length >= 4) { + sentinelBounds = body.bbox; + layerOpts.bounds = L.latLngBounds( + [body.bbox[1], body.bbox[0]], + [body.bbox[3], body.bbox[2]], + ); + } else { + sentinelBounds = null; + } + sentinelLayer = dropLayer(sentinelLayer); + sentinelLayer = L.tileLayer(body.tileUrl, layerOpts).addTo(map); + sentinelItemId = body.itemId || null; if (countEl) countEl.textContent = String(body.polarization || 'SAR').toUpperCase(); if (hint) { const when = body.datetime ? ` · ${body.datetime}` : ''; @@ -3477,17 +3516,44 @@ async function onGfClose(e) { toggleGeofenceDraw(); loadGeofences(); } +async function deleteGeofence(id) { + if (!id) return; + try { + const r = await fetch(`${API}/api/geofences/${encodeURIComponent(id)}`, { method: 'DELETE' }); + if (!r.ok && r.status !== 204) throw new Error('delete failed'); + } catch (err) { console.error('geofence delete failed', err); } + loadGeofences(); +} async function loadGeofences() { if (!map) return; try { const r = await overlayFetch(`${API}/api/geofences`); const rows = await r.json(); + const list = Array.isArray(rows) ? rows : []; + const box = document.getElementById('gf-list'); + if (box) { + box.innerHTML = list.map(f => { + const id = newsEsc(f.id || ''); + const name = newsEsc(f.name || 'Fence'); + return `
${name}` + + `
`; + }).join(''); + } if (gfSaved && map.hasLayer(gfSaved)) map.removeLayer(gfSaved); - const feats = (Array.isArray(rows) ? rows : []).map(f => ({ + const feats = list.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 }, + onEachFeature: (feat, layer) => { + const id = feat.properties && feat.properties.id; + const name = newsEsc((feat.properties && feat.properties.name) || 'Fence'); + if (!id) return; + layer.bindPopup( + `
${name}
` + + `
` + ); + }, }).addTo(map); } catch (e) { if (!isAbort(e)) console.error('geofences load failed', e); } } diff --git a/app/tracks.py b/app/tracks.py index b3a3d83..a31a7c5 100644 --- a/app/tracks.py +++ b/app/tracks.py @@ -186,15 +186,14 @@ async def track_range() -> dict: 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 + SELECT min(t) AS tmin, max(t) AS tmax FROM ( + SELECT min(bucket) AS t FROM vessel_tracks_1min + UNION ALL SELECT max(bucket) FROM vessel_tracks_1min + UNION ALL SELECT min(bucket) FROM aircraft_tracks_1min + UNION ALL SELECT max(bucket) FROM aircraft_tracks_1min + UNION ALL SELECT min(poll_at) FROM vessels + UNION ALL SELECT max(poll_at) FROM vessels + ) s """ ))).mappings().first() if not row or row["tmin"] is None: diff --git a/app/vesselapi.py b/app/vesselapi.py index 2c41612..0b2317b 100644 --- a/app/vesselapi.py +++ b/app/vesselapi.py @@ -17,6 +17,7 @@ from __future__ import annotations import asyncio import calendar +import json import logging import os from datetime import date, datetime, timezone @@ -32,7 +33,7 @@ from config import ( VESSELAPI_MAX_CALLS_PER_DAY, ) from database import async_session, engine, metadata -from live_layers import to_marker, upsert_vessel +from live_layers import parse_bbox, to_marker, upsert_vessel, vessel_last_known, vessel_lock logger = logging.getLogger("osint.vesselapi") @@ -176,6 +177,49 @@ def transform_vesselapi_payload(payload: dict | None) -> list[dict]: return out +def utc_day_start(now: datetime) -> datetime: + """Floor ``now`` to 00:00:00 UTC.""" + if now.tzinfo is None: + now = now.replace(tzinfo=timezone.utc) + now = now.astimezone(timezone.utc) + return now.replace(hour=0, minute=0, second=0, microsecond=0) + + +def pick_poll_at(poll_times: list[datetime], as_of: datetime) -> datetime | None: + """Latest poll timestamp at or before ``as_of`` (DVR as-of).""" + if as_of.tzinfo is None: + as_of = as_of.replace(tzinfo=timezone.utc) + else: + as_of = as_of.astimezone(timezone.utc) + eligible: list[datetime] = [] + for raw in poll_times: + ts = raw if raw.tzinfo else raw.replace(tzinfo=timezone.utc) + ts = ts.astimezone(timezone.utc) + if ts <= as_of: + eligible.append(ts) + return max(eligible) if eligible else None + + +def snapshot_as_of(rows: list[dict], as_of: datetime) -> list[dict]: + """Keep only rows from the latest poll_at ≤ ``as_of``.""" + chosen = pick_poll_at( + [r["poll_at"] for r in rows if r.get("poll_at") is not None], + as_of, + ) + if chosen is None: + return [] + out = [] + for row in rows: + ts = row.get("poll_at") + if ts is None: + continue + if ts.tzinfo is None: + ts = ts.replace(tzinfo=timezone.utc) + if ts.astimezone(timezone.utc) == chosen: + out.append(row) + return out + + # ── Durable daily quota (Postgres, survives restarts) ───────────────────── # Mirrors keystore.api_keys: lazy CREATE TABLE IF NOT EXISTS, no alembic fork. @@ -261,6 +305,183 @@ class PgQuotaStore: return (int(existing) if existing else 0) + 1 +# ── Daily VesselAPI snapshots (DVR as-of + survive restarts) ────────────── +# Cleared at the UTC day boundary so the table holds today's 5 polls only. + +_CREATE_VESSELS_SQL = text( + """ + CREATE TABLE IF NOT EXISTS vessels ( + mmsi TEXT NOT NULL, + poll_at 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, poll_at) + ) + """ +) +_CREATE_VESSELS_POLL_IDX = text( + "CREATE INDEX IF NOT EXISTS ix_vessels_poll_at ON vessels (poll_at DESC)" +) +_CREATE_VESSELS_BBOX_IDX = text( + "CREATE INDEX IF NOT EXISTS ix_vessels_bbox ON vessels (lon, lat)" +) + +_vessels_lock = asyncio.Lock() +_vessels_ensured = False + + +async def ensure_vessels_table() -> None: + global _vessels_ensured + if _vessels_ensured: + return + async with _vessels_lock: + if _vessels_ensured: + return + async with engine.begin() as conn: + await conn.execute(_CREATE_VESSELS_SQL) + await conn.execute(_CREATE_VESSELS_POLL_IDX) + await conn.execute(_CREATE_VESSELS_BBOX_IDX) + _vessels_ensured = True + + +def _marker_from_vessel_row(r) -> dict: + extra = r.get("extra") or {} + if isinstance(extra, str): + try: + extra = json.loads(extra) + except (TypeError, ValueError): + extra = {} + if not isinstance(extra, dict): + extra = {} + extra.setdefault("src", "vesselapi") + poll_at = r.get("poll_at") + if poll_at is not None and hasattr(poll_at, "isoformat"): + extra["poll_at"] = poll_at.isoformat() + marker = to_marker( + str(r["id"]), r["lat"], r["lon"], + heading=r.get("heading"), speed=r.get("speed"), + label=r.get("label") or str(r["id"]), + extra=extra, + ) + marker["seen_at"] = extra.get("poll_at") or datetime.now(timezone.utc).isoformat() + return marker + + +async def persist_vessel_snapshot(markers: list[dict], poll_at: datetime) -> None: + """Write one VesselAPI poll into ``vessels`` (today's snapshots).""" + await ensure_vessels_table() + if not markers: + return + async with async_session() as session: + for m in markers: + vid = str(m.get("id") or "") + lat, lon = m.get("lat"), m.get("lon") + if not vid or lat is None or lon is None: + continue + extra = dict(m.get("extra") or {}) + extra.setdefault("src", "vesselapi") + await session.execute( + text( + """ + INSERT INTO vessels + (mmsi, poll_at, lat, lon, heading, speed, label, extra) + VALUES + (:mmsi, :poll_at, :lat, :lon, :heading, :speed, :label, + CAST(:extra AS jsonb)) + ON CONFLICT (mmsi, poll_at) DO UPDATE SET + lat = EXCLUDED.lat, + lon = EXCLUDED.lon, + heading = EXCLUDED.heading, + speed = EXCLUDED.speed, + label = EXCLUDED.label, + extra = EXCLUDED.extra + """ + ), + { + "mmsi": vid, + "poll_at": poll_at, + "lat": float(lat), + "lon": float(lon), + "heading": m.get("heading"), + "speed": m.get("speed"), + "label": m.get("label") or vid, + "extra": json.dumps(extra), + }, + ) + await session.commit() + + +async def purge_old_vessels(before: datetime | None = None) -> None: + """Drop snapshots from before the current UTC day (or ``before``).""" + await ensure_vessels_table() + cutoff = before or utc_day_start(datetime.now(timezone.utc)) + async with async_session() as session: + await session.execute( + text("DELETE FROM vessels WHERE poll_at < :cutoff"), + {"cutoff": cutoff}, + ) + await session.commit() + + +async def fetch_vessels_as_of( + ts: datetime, + bbox: str | None = None, + limit: int = 2000, +) -> list[dict]: + """Latest VesselAPI poll at or before ``ts`` (DVR as-of, not exact minute).""" + try: + await ensure_vessels_table() + async with async_session() as session: + poll = (await session.execute( + text("SELECT max(poll_at) FROM vessels WHERE poll_at <= :ts"), + {"ts": ts}, + )).scalar() + if poll is None: + return [] + sql = """ + SELECT mmsi AS id, lat, lon, heading, speed, label, extra, poll_at + FROM vessels + WHERE poll_at = :poll + """ + params: dict = {"poll": poll, "limit": limit} + if bbox: + minlon, minlat, maxlon, maxlat = parse_bbox(bbox) + sql += ( + " AND lon BETWEEN :minlon AND :maxlon" + " AND lat BETWEEN :minlat AND :maxlat" + ) + params.update( + minlon=minlon, minlat=minlat, maxlon=maxlon, maxlat=maxlat, + ) + sql += " LIMIT :limit" + rows = (await session.execute(text(sql), params)).mappings().all() + return [_marker_from_vessel_row(r) for r in rows] + except Exception: + logger.exception("VesselAPI snapshot fetch failed") + return [] + + +async def hydrate_last_known() -> int: + """Seed in-memory last-known from today's latest poll (app boot).""" + try: + rows = await fetch_vessels_as_of(datetime.now(timezone.utc)) + except Exception: + logger.exception("VesselAPI hydrate failed") + return 0 + if not rows: + return 0 + async with vessel_lock: + for m in rows: + vid = str(m.get("id") or "") + if vid: + vessel_last_known[vid] = m + return len(rows) + + # ── Budget / scheduling (pure, unit-testable) ───────────────────────────── def days_left_in_month(now: datetime) -> int: @@ -406,6 +627,11 @@ async def poll_once(store, boxes: list[tuple[float, float, float, float]], key: markers = transform_vesselapi_payload(data) for m in markers: await upsert_vessel(m) + try: + await persist_vessel_snapshot(markers, now) + await purge_old_vessels(utc_day_start(now)) + except Exception: # noqa: BLE001 — live overlay must not die on persist + logger.exception("VesselAPI snapshot persist failed") logger.info( "VesselAPI poll OK: %d vessels (remaining=%s, calls_today=%d)", len(markers), remaining, calls, diff --git a/docs/news.md b/docs/news.md index d94d29f..d64b31d 100644 --- a/docs/news.md +++ b/docs/news.md @@ -128,11 +128,12 @@ Key set **unchanged** (no `lat`/`lon` on articles; geo lives on `/api/news/map`) `?kind=daily_recap` pins the nightly 24h recap. Empty DB → `[]` (no crash). Malformed `kind` → `422`. -### GET /api/news/ticker — flagged HUD headlines +### GET /api/news/ticker — HUD headlines -Critical/high `news_items` with `kind=ticker` only. Do **not** reuse -`GET /api/alerts`. Bottom HUD `#nt-track` scrolls these rows, not a dump of -the whole brief. +Critical/high `news_items` with `kind=ticker` first. If none are flagged, +medium/low ticker rows fill the tape so the dock is not blank. Do **not** +reuse `GET /api/alerts`. Bottom HUD `#nt-track` scrolls these rows, not a +dump of the whole brief. | Query param | Meaning | Default | |---|---|---| @@ -226,10 +227,10 @@ markdown json fences, then brace-slices: } ``` -Persist ticker/map only for `importance` in `critical`/`high`. Map rows also -need valid coords; Unknown / invented places are dropped. Caps: 12 ticker -(≤140 chars, no markdown), 20 map. Empty ticker is allowed. `summary_en` -lands in `article_summaries.summary_text`. +Persist ticker for critical/high first; if none, persist medium/low so the +tape is not empty. Map rows stay critical/high with valid coords; Unknown / +invented places are dropped. Caps: 12 ticker (≤140 chars, no markdown), 20 +map. `summary_en` lands in `article_summaries.summary_text`. ## Configuration (all via env / `.env`) diff --git a/news/summerizer/intel.py b/news/summerizer/intel.py index d404f44..1f9874c 100644 --- a/news/summerizer/intel.py +++ b/news/summerizer/intel.py @@ -7,6 +7,7 @@ import re _EMPTY = {"summary_en": "", "ticker": [], "map_items": []} _KEEP = frozenset({"critical", "high"}) +_RANK = {"critical": 0, "high": 1, "medium": 2, "low": 3} _THINK_RE = re.compile(r".*?", re.DOTALL) _FENCE_RE = re.compile(r"```(?:json)?", re.IGNORECASE) @@ -58,19 +59,29 @@ def _trimmed_headline(row: dict, limit: int) -> str: def select_ticker(rows: list) -> list: - out = [] + flagged = [] + medium = [] + low = [] for row in rows: - if row.get("importance") not in _KEEP: + imp = row.get("importance") + if imp not in _RANK: continue headline = _trimmed_headline(row, TICKER_HEADLINE_MAX) if not headline: continue item = dict(row) item["headline"] = headline - out.append(item) - if len(out) >= TICKER_CAP: + if imp in _KEEP: + flagged.append(item) + elif imp == "medium": + medium.append(item) + else: + low.append(item) + if len(flagged) >= TICKER_CAP: break - return out + if flagged: + return flagged[:TICKER_CAP] + return (medium + low)[:TICKER_CAP] def select_map(items: list) -> list: diff --git a/news/summerizer/nous_client.py b/news/summerizer/nous_client.py index da0b2c4..fc480f1 100644 --- a/news/summerizer/nous_client.py +++ b/news/summerizer/nous_client.py @@ -8,6 +8,10 @@ import httpx _DEFAULT_UA = "osint-dashboard-news-summarizer" _DEFAULT_BASE = "https://inference-api.nousresearch.com/v1" +_JSON_SYSTEM = ( + "You are an OSINT executive briefer. Reply with a single complete JSON object. " + "Never truncate mid-sentence. If you run out of room, drop the lowest-priority item." +) def chat(prompt, *, api_key, model, base_url, json_mode=False) -> str: @@ -17,20 +21,37 @@ def chat(prompt, *, api_key, model, base_url, json_mode=False) -> str: "Authorization": f"Bearer {api_key}", "User-Agent": os.environ.get("OSINT_USER_AGENT") or _DEFAULT_UA, } + max_tokens = 8192 if json_mode else 4096 + timeout = 120.0 if json_mode else 60.0 + messages = [{"role": "user", "content": prompt}] + if json_mode: + messages = [ + {"role": "system", "content": _JSON_SYSTEM}, + {"role": "user", "content": prompt}, + ] payload = { "model": model, - "messages": [{"role": "user", "content": prompt}], + "messages": messages, "temperature": 0.2, - "max_tokens": 4096, + "max_tokens": max_tokens, } if json_mode: payload["response_format"] = {"type": "json_object"} + last_content = "" try: - with httpx.Client(timeout=60.0) as client: - resp = client.post(url, headers=headers, json=payload) - if resp.status_code == 401 or resp.status_code >= 500: - return "" - data = resp.json() - return data["choices"][0]["message"]["content"] + for attempt in range(2): + with httpx.Client(timeout=timeout) as client: + resp = client.post(url, headers=headers, json=payload) + if resp.status_code == 401 or resp.status_code >= 500: + return "" + data = resp.json() + choice = (data.get("choices") or [{}])[0] + last_content = (choice.get("message") or {}).get("content") or "" + finish = choice.get("finish_reason") + if finish == "length" and attempt == 0: + payload["max_tokens"] = min(int(payload["max_tokens"]) * 2, 16384) + continue + return last_content + return last_content except Exception: return "" diff --git a/news/summerizer/summarizer.py b/news/summerizer/summarizer.py index 73f9cb6..acc1262 100644 --- a/news/summerizer/summarizer.py +++ b/news/summerizer/summarizer.py @@ -94,7 +94,7 @@ FUTURES_TICKERS = { MAP_PROMPT_DEFAULT = """\ You are a precise, factual OSINT news processor. Your ONLY source of information is the articles provided below. Do NOT add external knowledge, assumptions, training data, or invented facts. -Focus on breaking important news (geopolitical, military/conflict, security, disasters, major political developments). Ignore futures prices, commodity tape, ticker chatter, and routine market moves unless they themselves are the breaking event. +Focus on breaking important news (geopolitical, military/conflict, security, disasters, major political developments). Ignore futures prices, commodity tape, ticker chatter, and routine market moves unless they themselves are the breaking event. If the batch has no critical/high stories, still extract minor incidents and crime reports. Write every field in English. Translate if the article is not English. @@ -133,7 +133,11 @@ You are writing an English operator HUD brief from the article facts in DATA bel Always write a real summary_en that recaps the most important stories present in DATA. Rank geopolitics, military/conflict, security, disasters, and major political developments first. Ignore futures prices, commodity tape, ticker chatter, and routine market data — do not treat price ticks as news. -ticker and map_items may be empty if nothing is critical or high. Never replace summary_en with a canned empty-brief sentence when DATA contains article facts. +Lead with critical and high breaking events. If DATA has no critical/high stories, fill the brief with minor incidents and crime reports rather than writing an empty or unfinished brief. Never truncate mid-sentence; finish every sentence. If you run out of room, drop the lowest-priority item instead of cutting a line short. + +ticker: prefer critical and high. If nothing is critical or high, fill ticker with medium then low incidents and crime so the HUD is not blank. + +map_items may be empty if no located critical/high event is explicit in the data. Demand a single JSON object (no markdown fences) with this exact shape: @@ -143,9 +147,9 @@ Demand a single JSON object (no markdown fences) with this exact shape: "map_items": [{"headline": "", "importance": "critical", "location_name": "", "lat": 0, "lon": 0, "location_confidence": "city", "category": "military/conflict", "url": ""}] } -ticker: only critical and high, max 12, ≤140 chars, no markdown. -map_items: only critical and high where a real-world location is explicit in the data. Estimate lat/lon. If location is Unknown or not in the data, omit the item. Never invent a place. Max 20. -summary_en: English markdown brief of breaking important news for an operator HUD (bullets or short paragraphs). Cover the actual stories in DATA. +ticker: max 12, ≤140 chars, no markdown. Rank critical > high > medium > low. +map_items: only where a real-world location is explicit in the data. Estimate lat/lon. If location is Unknown or not in the data, omit the item. Never invent a place. Max 20. +summary_en: English markdown executive brief for an operator HUD (4–8 complete bullets or short paragraphs). Cover the actual stories in DATA. Complete — never an unfinished sentence. DATA: {final_input} @@ -156,7 +160,9 @@ You are writing a daily recap of the last 24 hours of news for an OSINT operator Always write a real summary_en daily recap of the most important stories in DATA. Rank geopolitics, military/conflict, security, disasters, and major political developments first. Ignore futures prices, commodity tape, ticker chatter, and routine market data — do not treat price ticks as news. -ticker and map_items may be empty if nothing is critical or high. Never replace summary_en with a canned empty-brief sentence when DATA contains article facts. +Lead with critical and high breaking events. If DATA has no critical/high stories, fill the recap with minor incidents and crime reports rather than writing an empty or unfinished recap. Never truncate mid-sentence; finish every sentence. + +ticker: prefer critical and high. If nothing is critical or high, fill ticker with medium then low incidents and crime so the HUD is not blank. Demand a single JSON object (no markdown fences) with this exact shape: @@ -166,9 +172,9 @@ Demand a single JSON object (no markdown fences) with this exact shape: "map_items": [{"headline": "", "importance": "critical", "location_name": "", "lat": 0, "lon": 0, "location_confidence": "city", "category": "military/conflict", "url": ""}] } -ticker: only critical and high, max 12, ≤140 chars, no markdown. -map_items: only critical and high where a real-world location is explicit in the data. Estimate lat/lon. If location is Unknown or not in the data, omit the item. Never invent a place. Max 20. -summary_en: English markdown daily recap of the last 24 hours of breaking important news. Cover the actual stories in DATA. +ticker: max 12, ≤140 chars, no markdown. Rank critical > high > medium > low. +map_items: only where a real-world location is explicit in the data. Estimate lat/lon. If location is Unknown or not in the data, omit the item. Never invent a place. Max 20. +summary_en: English markdown daily recap of the last 24 hours. Complete sentences. Cover the actual stories in DATA. DATA: {final_input} diff --git a/news/summerizer/tests/test_intel.py b/news/summerizer/tests/test_intel.py index dde2b5c..02446ea 100644 --- a/news/summerizer/tests/test_intel.py +++ b/news/summerizer/tests/test_intel.py @@ -32,6 +32,16 @@ def test_select_ticker_keeps_critical_high_caps_12(): assert len(out) == 12 assert all(r["importance"] in ("critical", "high") for r in out) + +def test_select_ticker_falls_back_to_medium_low_when_nothing_flagged(): + rows = [ + {"headline": "shop theft", "importance": "low"}, + {"headline": "highway crash", "importance": "medium"}, + {"headline": "none", "importance": "none"}, + ] + out = select_ticker(rows) + assert [r["headline"] for r in out] == ["highway crash", "shop theft"] + def test_select_map_requires_valid_coords_and_flag(): items = [ {"headline": "A", "importance": "critical", "lat": 50.45, "lon": 30.52, "location_name": "Kyiv"}, diff --git a/news/summerizer/tests/test_nous_client.py b/news/summerizer/tests/test_nous_client.py index 002faa4..848db14 100644 --- a/news/summerizer/tests/test_nous_client.py +++ b/news/summerizer/tests/test_nous_client.py @@ -67,6 +67,33 @@ def test_json_mode_sets_response_format(monkeypatch): captured = _install_fake(monkeypatch, lambda *a: _ok_response("{}")) chat("p", api_key="k", model="m", base_url=BASE, json_mode=True) assert captured["json"]["response_format"] == {"type": "json_object"} + assert captured["json"]["max_tokens"] >= 8192 + roles = [m["role"] for m in captured["json"]["messages"]] + assert "system" in roles + assert "user" in roles + + +def test_retries_once_when_finish_reason_is_length(monkeypatch): + calls = {"n": 0} + + def post_impl(*a): + calls["n"] += 1 + if calls["n"] == 1: + resp = MagicMock() + resp.status_code = 200 + resp.json.return_value = { + "choices": [{ + "message": {"content": "{\"summary_en\": \"cut off"}, + "finish_reason": "length", + }] + } + return resp + return _ok_response('{"summary_en": "complete brief."}') + + _install_fake(monkeypatch, post_impl) + out = chat("p", api_key="k", model="m", base_url=BASE, json_mode=True) + assert calls["n"] == 2 + assert "complete brief" in out def test_401_returns_empty_string(monkeypatch): diff --git a/news/summerizer/tests/test_prompts.py b/news/summerizer/tests/test_prompts.py index e5d4374..954e5b3 100644 --- a/news/summerizer/tests/test_prompts.py +++ b/news/summerizer/tests/test_prompts.py @@ -22,6 +22,14 @@ def test_summary_prompt_focuses_on_breaking_news_not_futures(): assert "commodity" in p or "market" in p +def test_summary_prompt_covers_critical_then_incidents(): + p = SUMMARY_PROMPT_DEFAULT.lower() + assert "critical" in p + assert "crime" in p + assert "incident" in p + assert "complete" in p or "truncat" in p or "unfinished" in p or "mid-sentence" in p + + def test_summary_prompt_does_not_bail_out_with_canned_empty_brief(): p = SUMMARY_PROMPT_DEFAULT assert "AND STOP" not in p diff --git a/tests/test_api_news.py b/tests/test_api_news.py index 0ba569d..e94b0e3 100644 --- a/tests/test_api_news.py +++ b/tests/test_api_news.py @@ -230,6 +230,21 @@ def test_api_news_ticker_returns_only_flagged(clean_news): assert item["url"] == "https://example.com/ticker" +@requires_db +def test_api_news_ticker_falls_back_to_lesser_when_nothing_flagged(clean_news): + sid = _seed_summary("quiet brief", "2026-08-27T18:05:00+00:00", "Hermes-4.3-36B") + _seed_news_item( + sid, "ticker", "Shop theft downtown", "low", + location_name="Raleigh", url="https://example.com/theft", + ) + resp = _get("/api/news/ticker") + assert resp.status_code == 200 + body = resp.json() + assert len(body) == 1 + assert body[0]["headline"] == "Shop theft downtown" + assert body[0]["importance"] == "low" + + @requires_db def test_api_news_map_returns_only_flagged_with_coords(clean_news): _seed_flagged_items() diff --git a/tests/test_frontend_reliability.py b/tests/test_frontend_reliability.py index d734f49..2f6dbde 100644 --- a/tests/test_frontend_reliability.py +++ b/tests/test_frontend_reliability.py @@ -72,6 +72,14 @@ def test_chokepoint_skips_aisstream_subscribe_outside_conus(): assert "minlat,minlon,maxlat,maxlon" in HTML.split("function chokepointLeafletBounds")[1][:400] +def test_news_ticker_polls_more_often_than_summarizer_cycle(): + assert "NEWS_REFRESH_MS" in HTML + # Summarizer is 15 min; ticker should refresh on a shorter cadence so + # lesser-news fills show up without waiting for the next brief. + line = [ln for ln in HTML.splitlines() if "NEWS_REFRESH_MS" in ln][0] + assert "900000" not in line + + def test_phone_chokepoints_use_select_not_buttons(): mobile = HTML.split("@media (max-width: 820px)")[1].split("@media (prefers-reduced-motion")[0] assert "#chokepoint-select { display: block; }" in mobile diff --git a/tests/test_geofence_frontend.py b/tests/test_geofence_frontend.py new file mode 100644 index 0000000..50ee03e --- /dev/null +++ b/tests/test_geofence_frontend.py @@ -0,0 +1,26 @@ +"""Geofence layer panel: draw + delete (DELETE /api/geofences/{id}).""" + +from __future__ import annotations + +from pathlib import Path + +ROOT = Path(__file__).resolve().parent.parent +HTML = (ROOT / "app/static/index.html").read_text() + + +def test_geofence_panel_has_list_and_delete_hook(): + assert 'id="gf-draw"' in HTML + assert 'id="gf-list"' in HTML + assert "function deleteGeofence" in HTML + assert "method: 'DELETE'" in HTML or 'method: "DELETE"' in HTML + assert "/api/geofences/" in HTML + + +def test_load_geofences_renders_delete_controls(): + js = HTML.split("async function loadGeofences", 1)[1].split( + "async function loadFireAircraftHits", 1 + )[0] + assert "gf-list" in js + assert "deleteGeofence" in js + assert "onEachFeature" in js + assert "bindPopup" in js diff --git a/tests/test_live_layers.py b/tests/test_live_layers.py index 69f8164..ae0f68e 100644 --- a/tests/test_live_layers.py +++ b/tests/test_live_layers.py @@ -7,6 +7,7 @@ from live_layers import ( filter_points_bbox, parse_bbox, quantize_bbox, + pick_sentinel_feature, rainviewer_tile_url, sign_cog_url, sentinel1_tile_url, @@ -734,8 +735,9 @@ def test_fetch_sentinel1_vv_signed_tile_url(monkeypatch): post_url, post_json = calls[0][1], calls[0][2] assert post_url.endswith("/api/stac/v1/search") assert post_json["collections"] == ["sentinel-1-grd"] - assert post_json["limit"] == 1 + assert post_json["limit"] >= 1 assert post_json["sortby"][0]["direction"] == "desc" + assert "bbox" in out def test_fetch_sentinel1_uses_hh_when_vv_missing(monkeypatch): @@ -783,3 +785,22 @@ def test_fetch_sentinel1_none_when_no_vv_or_hh(monkeypatch): _cache.clear() assert asyncio.run(fetch_sentinel1("-80,35,-79,36")) is None + + +def test_pick_sentinel_feature_prefers_scene_covering_center(): + features = [ + {"id": "far", "bbox": [10.0, 10.0, 12.0, 12.0]}, + {"id": "cover", "bbox": [-80.5, 34.5, -78.5, 36.5]}, + {"id": "also-far", "bbox": [-10.0, 0.0, -8.0, 2.0]}, + ] + picked = pick_sentinel_feature(features, -79.5, 35.5) + assert picked["id"] == "cover" + + +def test_pick_sentinel_feature_falls_back_to_first_when_none_cover(): + features = [ + {"id": "a", "bbox": [10.0, 10.0, 12.0, 12.0]}, + {"id": "b", "bbox": [20.0, 20.0, 22.0, 22.0]}, + ] + assert pick_sentinel_feature(features, -79.5, 35.5)["id"] == "a" + assert pick_sentinel_feature([], -79.5, 35.5) is None diff --git a/tests/test_sentinel1_frontend.py b/tests/test_sentinel1_frontend.py index 94a30a8..fdc1d6b 100644 --- a/tests/test_sentinel1_frontend.py +++ b/tests/test_sentinel1_frontend.py @@ -39,3 +39,11 @@ def test_sentinel1_not_fetched_on_init_unless_on(): assert "loadSentinel1()" not in init refresh = HTML.split("function refreshLiveOverlays", 1)[1].split("function addExtraAttrib", 1)[0] assert "if (sentinelOn) loadSentinel1();" in refresh + + +def test_sentinel1_reuses_covering_scene_and_clips_tiles(): + js = HTML.split("async function loadSentinel1", 1)[1].split("function loadThermal", 1)[0] + assert "sentinelStillCovers" in HTML + assert "itemId" in js + assert "L.latLngBounds" in js + assert "sentinelBounds" in HTML diff --git a/tests/test_vesselapi.py b/tests/test_vesselapi.py index f7ec9c1..de5b574 100644 --- a/tests/test_vesselapi.py +++ b/tests/test_vesselapi.py @@ -234,6 +234,8 @@ def _patch_side_effects(monkeypatch): monkeypatch.setattr("tracks.record_position", _noop) monkeypatch.setattr("geofence.record_and_notify", _noop) + monkeypatch.setattr(vesselapi, "persist_vessel_snapshot", _noop) + monkeypatch.setattr(vesselapi, "purge_old_vessels", _noop) def test_poll_once_lands_markers_in_vessel_last_known(monkeypatch): diff --git a/tests/test_vessels_snapshot.py b/tests/test_vessels_snapshot.py new file mode 100644 index 0000000..a9ea4ea --- /dev/null +++ b/tests/test_vessels_snapshot.py @@ -0,0 +1,36 @@ +"""VesselAPI daily snapshot store — as-of DVR + UTC-day purge (no DB).""" + +from __future__ import annotations + +from datetime import datetime, timezone + +from vesselapi import pick_poll_at, snapshot_as_of, utc_day_start + + +def test_utc_day_start_floors_to_midnight_utc(): + now = datetime(2026, 8, 29, 15, 30, 12, tzinfo=timezone.utc) + assert utc_day_start(now) == datetime(2026, 8, 29, 0, 0, tzinfo=timezone.utc) + + +def test_pick_poll_at_returns_latest_snapshot_at_or_before_as_of(): + t1 = datetime(2026, 8, 29, 0, 0, tzinfo=timezone.utc) + t2 = datetime(2026, 8, 29, 4, 48, tzinfo=timezone.utc) + t3 = datetime(2026, 8, 29, 9, 36, tzinfo=timezone.utc) + as_of = datetime(2026, 8, 29, 6, 0, tzinfo=timezone.utc) + assert pick_poll_at([t1, t2, t3], as_of) == t2 + assert pick_poll_at([t1, t2, t3], t1) == t1 + assert pick_poll_at([t1, t2, t3], datetime(2026, 8, 28, 23, tzinfo=timezone.utc)) is None + + +def test_snapshot_as_of_returns_the_matching_poll_only(): + t1 = datetime(2026, 8, 29, 0, 0, tzinfo=timezone.utc) + t2 = datetime(2026, 8, 29, 4, 48, tzinfo=timezone.utc) + rows = [ + {"id": "1", "poll_at": t1, "lat": 26.5, "lon": 56.0}, + {"id": "2", "poll_at": t1, "lat": 26.6, "lon": 56.1}, + {"id": "1", "poll_at": t2, "lat": 26.7, "lon": 56.2}, + ] + out = snapshot_as_of(rows, datetime(2026, 8, 29, 6, 0, tzinfo=timezone.utc)) + assert {r["id"] for r in out} == {"1"} + assert out[0]["lat"] == 26.7 + assert all(r["poll_at"] == t2 for r in out)