mirror of
https://github.com/zvx-echo6/meshai.git
synced 2026-06-10 17:04:45 +02:00
feat(v0.5.8b): persistence foundation + WFIGS handler + universal cold-start grace
Three integrated pieces that ship together because they were designed as one safety story: (1) PERSISTENCE FOUNDATION -- new meshai/persistence/ module with SQLite db.py, schema migration framework (v1), 13 tables covering all adapter event shapes (traffic_events, fires, firms_pixels, quake_events, nws_alerts, gauge_readings, swpc_events) + mesh state (mesh_nodes, mesh_telemetry, mesh_positions, mesh_messages_in, mesh_broadcasts_out, mesh_health_events) + cross-cutting event_log + schema_meta. WAL mode for reader concurrency, single-writer pattern, MESHAI_DB_PATH env var, mounted at /data/meshai.sqlite via existing docker-compose meshai_data volume. .gitignore updated. (2) WFIGS HANDLER -- meshai/central/wfigs_handler.py implements the first per-adapter handler that uses the persistence layer. Format: MEDIUM style with town/landclass/county fallback chain, lat/lon at 3-decimal precision, New:/Update: prefix. 8h-rate-limited change-detection per IRWIN via fires.last_broadcast_at. Skips tombstones and perimeters silently (logged to event_log with handled=0). Acres fallback chain DailyAcres -> IncidentSize -> raw.DiscoveryAcres -> raw.FinalAcres -> N/A. Pass-through Initial Attack auto-numbered names (IA 1, IA 2). (3) UNIVERSAL COLD-START GRACE -- meshai/notifications/pipeline/dispatcher.py grows a configurable grace window (cold_start_grace_seconds, default 60s, GUI-editable per Rule 17). Anchored to first-event-seen (not container boot), so the grace activates the moment broadcasts could fire. Suppresses mesh delivery during the window; handler-side persistence (fires UPSERT, event_log) still happens normally. New _cold_start_dropped counter exposed in dispatch_stats(). Designed to protect against JetStream backlog spam at toggle-flip time, applies universally to ALL adapters. (4) WFIGS HANDLER CALLBACK REFACTOR -- New:/Update: prefix now keys on fires.last_broadcast_at IS NULL (not row-missing), and last_broadcast_* field updates moved to a post-broadcast commit callback that the dispatcher invokes ONLY on successful delivery. This means: cold-start-suppressed events leave fires.last_broadcast_at NULL, so when they eventually broadcast post-grace, they correctly render as New: (first ACTUAL delivery for that IRWIN), not Update:. event_log.handled and mesh_broadcasts_out audit row also gated on the same callback -- decoupling persistence rows from broadcast rows for an honest audit trail. New tests: 15 in test_wfigs_handler.py, 15 in test_persistence.py, additional cold-start grace tests in test_dispatcher.py (+4 WFIGS callback scenarios). Synthetic probes wfigs-cleaned-samples.md (initial) and wfigs-cleaned-samples-v2.md (cold-start verification) generated against isolated temp SQLite databases. CT108 /data/meshai.sqlite untouched during build. Master stays off. No live toggle flips. Test count: was 535 (v0.5.7 baseline) -> 566 (persistence) -> 581 (wfigs handler) -> 589 expected (cold-start grace). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
7751a40c6c
commit
053d67db6e
16 changed files with 2652 additions and 1 deletions
3
.gitignore
vendored
3
.gitignore
vendored
|
|
@ -71,3 +71,6 @@ Thumbs.db
|
|||
!.env.example
|
||||
local.yaml
|
||||
!local.yaml.example
|
||||
data/*.sqlite
|
||||
data/*.sqlite-wal
|
||||
data/*.sqlite-shm
|
||||
|
|
|
|||
|
|
@ -64,6 +64,7 @@ interface NotificationsConfig {
|
|||
quiet_hours_enabled: boolean
|
||||
quiet_hours_start: string
|
||||
quiet_hours_end: string
|
||||
cold_start_grace_seconds?: number
|
||||
rules: NotificationRuleConfig[]
|
||||
toggles?: Record<string, NotificationToggle>
|
||||
}
|
||||
|
|
@ -2118,6 +2119,22 @@ export default function Notifications() {
|
|||
)}
|
||||
</div>
|
||||
|
||||
{/* Cold-start grace -- v0.5.8b */}
|
||||
<div className="space-y-3 p-4 bg-[#0a0e17] rounded-lg border border-[#1e2a3a]">
|
||||
<div className="flex items-center gap-2">
|
||||
<label className="text-xs text-slate-500 uppercase tracking-wide">Cold-start grace</label>
|
||||
</div>
|
||||
<NumberInput
|
||||
label="Grace period (seconds)"
|
||||
value={config.cold_start_grace_seconds ?? 60}
|
||||
onChange={(v) => setConfig({ ...config, cold_start_grace_seconds: v })}
|
||||
min={0}
|
||||
max={600}
|
||||
helper="Suppress broadcasts for this many seconds after the first event arrives"
|
||||
info="When meshai starts seeing events for the first time, suppress mesh broadcasts for this many seconds to absorb any JetStream backlog. Persistence rows still get written; only broadcasts are suppressed."
|
||||
/>
|
||||
</div>
|
||||
|
||||
{/* Master Toggles */}
|
||||
{config.toggles && (
|
||||
<MasterToggles
|
||||
|
|
|
|||
|
|
@ -422,7 +422,15 @@ class CentralConsumer:
|
|||
from meshai.central_normalizer import normalize as _norm_envelope
|
||||
from meshai.notifications.renderers.work_zone import format_work_zone_mesh
|
||||
n = _norm_envelope(envelope)
|
||||
if n is not None and category in ("work_zone", "road_closure", "road_incident"):
|
||||
# v0.5.8 wfigs_handler dispatch -- WFIGS events route through
|
||||
# the persistence-backed change-detection handler (which also
|
||||
# logs to event_log for tombstones + perimeters). Other adapters
|
||||
# with a normalized dict (state_511_atis, wzdx) flow through the
|
||||
# work_zone renderer as before.
|
||||
if n is not None and str(n.get("_kind", "")).startswith("wfigs"):
|
||||
from meshai.central.wfigs_handler import handle_wfigs
|
||||
synthesized = handle_wfigs(n, envelope, subject, data=data) or None
|
||||
elif n is not None and category in ("work_zone", "road_closure", "road_incident"):
|
||||
synthesized = format_work_zone_mesh(n) or None
|
||||
except Exception:
|
||||
logger.exception("normalizer/renderer failed for adapter=%s category=%s",
|
||||
|
|
|
|||
346
meshai/central/wfigs_handler.py
Normal file
346
meshai/central/wfigs_handler.py
Normal file
|
|
@ -0,0 +1,346 @@
|
|||
"""WFIGS handler: persistence-backed change-detection + wire renderer.
|
||||
|
||||
v0.5.8b refactor: New: vs Update: decision now keys on `last_broadcast_at`,
|
||||
not on row existence. Cold-start scenarios where the dispatcher drops the
|
||||
broadcast (cold-start grace, stale filter, cooldown, dedup) leave the fires
|
||||
row with NULL last_broadcast_at, so the NEXT successful broadcast still
|
||||
gets the "New:" prefix -- it really is the first delivery for that fire.
|
||||
|
||||
Cases (resolved at handler entry):
|
||||
(i) row missing -> INSERT, prefix="New", return wire
|
||||
(ii) row exists, last_broadcast_at IS NULL
|
||||
-> UPDATE current_*, prefix="New",
|
||||
return wire (never broadcast yet)
|
||||
(iii) row exists, last_broadcast_at NOT NULL
|
||||
-> UPDATE current_*, gate on change +
|
||||
8h cooldown. If pass: prefix="Update",
|
||||
return wire; else return None.
|
||||
|
||||
The last_broadcast_* UPDATE has moved OUT of the handler and INTO a callback
|
||||
attached to event.data["_on_broadcast_committed"]. The dispatcher calls it
|
||||
ONLY after a successful broadcast. The mesh_broadcasts_out audit row is now
|
||||
inserted by the dispatcher (via event.data["_broadcast_audit"]) for the same
|
||||
reason -- it should only exist for actually-delivered broadcasts.
|
||||
|
||||
Concurrency: each consumer thread gets its own SQLite connection via
|
||||
meshai.persistence.get_db() (threading.local pool). Writes are serial
|
||||
inside that connection's autocommit mode.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import time
|
||||
from typing import Any, Optional
|
||||
|
||||
from meshai.persistence import get_db
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Broadcast cooldown per fire (8h). Without this, every poll cycle would
|
||||
# re-broadcast even when nothing changed. Spec: change-detection on acres
|
||||
# OR containment AND >=28800s since last broadcast.
|
||||
WFIGS_BROADCAST_COOLDOWN_S = 8 * 60 * 60 # 28800
|
||||
|
||||
|
||||
def _now() -> int:
|
||||
return int(time.time())
|
||||
|
||||
|
||||
# ---------- public entry --------------------------------------------------
|
||||
|
||||
|
||||
def handle_wfigs(normalized: dict, envelope: dict, subject: str,
|
||||
data: Optional[dict] = None,
|
||||
now: Optional[int] = None) -> Optional[str]:
|
||||
"""Route a normalized WFIGS dict through persistence + change-detection.
|
||||
|
||||
`data` is the mutable dict the caller (consumer._normalize) is composing
|
||||
into the Event. When a broadcast should fire, the handler attaches an
|
||||
`_on_broadcast_committed` callback and `_broadcast_audit` descriptor to
|
||||
it; the dispatcher invokes both AFTER a successful deliver().
|
||||
|
||||
Returns a wire string when a broadcast should fire, None otherwise.
|
||||
"""
|
||||
if not isinstance(normalized, dict):
|
||||
return None
|
||||
kind = normalized.get("_kind")
|
||||
if kind not in ("wfigs_incident", "wfigs_tombstone", "wfigs_perimeter"):
|
||||
return None
|
||||
|
||||
now = now if now is not None else _now()
|
||||
inner = envelope.get("data") or {} if isinstance(envelope, dict) else {}
|
||||
category = inner.get("category") or ""
|
||||
severity_word = _coerce_severity(inner.get("severity"))
|
||||
irwin_id = normalized.get("irwin_id")
|
||||
|
||||
try:
|
||||
conn = get_db()
|
||||
except Exception:
|
||||
logger.exception("wfigs_handler: persistence unavailable; "
|
||||
"deferring to default pipeline")
|
||||
return None
|
||||
|
||||
if kind in ("wfigs_tombstone", "wfigs_perimeter"):
|
||||
source = "wfigs_incidents" if kind == "wfigs_tombstone" else "wfigs_perimeters"
|
||||
_log_event(conn, now=now, source=source, category=category,
|
||||
severity_word=severity_word, irwin_id=irwin_id,
|
||||
subject=subject, handled=0,
|
||||
table_name=None, table_pk=irwin_id)
|
||||
return None
|
||||
|
||||
# ---- active incident ----
|
||||
# v0.5.8b: log handled=0 initially. The commit callback UPDATEs this
|
||||
# row to handled=1 if/when the dispatcher actually broadcasts -- if it
|
||||
# drops (cold-start grace, staleness, cooldown, dedup), the row stays
|
||||
# handled=0 and we can grep the event_log to find the suppressed events.
|
||||
log_id = _log_event_returning_id(
|
||||
conn, now=now, source="wfigs_incidents", category=category,
|
||||
severity_word=severity_word, irwin_id=irwin_id,
|
||||
subject=subject, handled=0,
|
||||
table_name="fires", table_pk=irwin_id)
|
||||
|
||||
row = conn.execute(
|
||||
"SELECT current_acres, current_contained_pct, last_broadcast_at, "
|
||||
"last_broadcast_acres, last_broadcast_contained "
|
||||
"FROM fires WHERE irwin_id = ?", (irwin_id,)).fetchone()
|
||||
|
||||
acres = normalized.get("acres")
|
||||
contained_pct = normalized.get("contained_pct")
|
||||
|
||||
# ---- (i) row missing -- INSERT, mark "New", but DO NOT set last_broadcast_*
|
||||
if row is None:
|
||||
conn.execute(
|
||||
"INSERT INTO fires(irwin_id, incident_name, incident_type, "
|
||||
"current_acres, current_contained_pct, status, lat, lon, "
|
||||
"county, state, landclass, declared_at, last_event_at, "
|
||||
"last_broadcast_at, last_broadcast_acres, last_broadcast_contained) "
|
||||
"VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
|
||||
(
|
||||
irwin_id,
|
||||
normalized.get("incident_name"),
|
||||
normalized.get("incident_type"),
|
||||
acres, contained_pct,
|
||||
None, # status reserved
|
||||
normalized.get("lat"), normalized.get("lon"),
|
||||
normalized.get("county"), normalized.get("state"),
|
||||
normalized.get("landclass"),
|
||||
normalized.get("declared_at_epoch"),
|
||||
now, # last_event_at
|
||||
None, None, None, # last_broadcast_* explicitly NULL
|
||||
),
|
||||
)
|
||||
wire = _render(normalized, prefix="New")
|
||||
_attach_commit_handles(data, irwin_id=irwin_id,
|
||||
acres=acres, contained_pct=contained_pct,
|
||||
event_log_row_id=log_id)
|
||||
return wire
|
||||
|
||||
# ---- (ii) row exists but never broadcast -- UPDATE current_*, prefix="New"
|
||||
if row["last_broadcast_at"] is None:
|
||||
conn.execute(
|
||||
"UPDATE fires SET current_acres=?, current_contained_pct=?, "
|
||||
"lat=COALESCE(?, lat), lon=COALESCE(?, lon), last_event_at=? "
|
||||
"WHERE irwin_id=?",
|
||||
(acres, contained_pct, normalized.get("lat"),
|
||||
normalized.get("lon"), now, irwin_id),
|
||||
)
|
||||
wire = _render(normalized, prefix="New")
|
||||
_attach_commit_handles(data, irwin_id=irwin_id,
|
||||
acres=acres, contained_pct=contained_pct,
|
||||
event_log_row_id=log_id)
|
||||
return wire
|
||||
|
||||
# ---- (iii) row exists AND already broadcast -- gate on change + 8h cooldown
|
||||
conn.execute(
|
||||
"UPDATE fires SET current_acres=?, current_contained_pct=?, "
|
||||
"lat=COALESCE(?, lat), lon=COALESCE(?, lon), last_event_at=? "
|
||||
"WHERE irwin_id=?",
|
||||
(acres, contained_pct, normalized.get("lat"),
|
||||
normalized.get("lon"), now, irwin_id),
|
||||
)
|
||||
|
||||
last_bcast_at = row["last_broadcast_at"]
|
||||
last_bcast_acres = row["last_broadcast_acres"]
|
||||
last_bcast_contained = row["last_broadcast_contained"]
|
||||
|
||||
# Forward-only change detection: more acres or higher containment counts.
|
||||
# Downward revisions and unchanged values do not warrant re-broadcast.
|
||||
changed_acres = (
|
||||
acres is not None
|
||||
and (last_bcast_acres is None or acres > last_bcast_acres)
|
||||
)
|
||||
changed_contained = (
|
||||
contained_pct is not None
|
||||
and (last_bcast_contained is None or contained_pct > last_bcast_contained)
|
||||
)
|
||||
eight_hours_passed = (
|
||||
last_bcast_at is None
|
||||
or (now - int(last_bcast_at) >= WFIGS_BROADCAST_COOLDOWN_S)
|
||||
)
|
||||
|
||||
if (changed_acres or changed_contained) and eight_hours_passed:
|
||||
wire = _render(normalized, prefix="Update")
|
||||
_attach_commit_handles(data, irwin_id=irwin_id,
|
||||
acres=acres, contained_pct=contained_pct,
|
||||
event_log_row_id=log_id)
|
||||
return wire
|
||||
|
||||
return None
|
||||
|
||||
|
||||
# ---------- commit-callback factory ---------------------------------------
|
||||
|
||||
|
||||
def _attach_commit_handles(data: Optional[dict], *, irwin_id: str,
|
||||
acres: Optional[float],
|
||||
contained_pct: Optional[int],
|
||||
event_log_row_id: Optional[int] = None) -> None:
|
||||
"""Attach `_on_broadcast_committed` callback + `_broadcast_audit`
|
||||
descriptor to the event-data dict. Both are read by the dispatcher
|
||||
AFTER a successful broadcast.
|
||||
|
||||
The callback closure captures the irwin_id + acres + contained_pct that
|
||||
triggered THIS broadcast. The dispatcher passes the actual delivery
|
||||
timestamp, which we record in last_broadcast_at. This keeps cold-start
|
||||
races correct: if the dispatcher drops the broadcast, the callback is
|
||||
not invoked and last_broadcast_at stays NULL -- so the NEXT successful
|
||||
broadcast still labels itself "New:".
|
||||
"""
|
||||
if not isinstance(data, dict):
|
||||
return
|
||||
|
||||
def _on_commit(committed_at: float) -> None:
|
||||
try:
|
||||
conn = get_db()
|
||||
except Exception:
|
||||
logger.exception(
|
||||
"wfigs commit callback: persistence unavailable; "
|
||||
"last_broadcast_* not updated for irwin=%s", irwin_id)
|
||||
return
|
||||
conn.execute(
|
||||
"UPDATE fires SET last_broadcast_at=?, last_broadcast_acres=?, "
|
||||
"last_broadcast_contained=? WHERE irwin_id=?",
|
||||
(int(committed_at), acres, contained_pct, irwin_id),
|
||||
)
|
||||
# Flip the matching event_log row to handled=1. A NULL row id
|
||||
# (caller forgot to thread it) is silently skipped -- the broadcast
|
||||
# still went out.
|
||||
if event_log_row_id is not None:
|
||||
conn.execute(
|
||||
"UPDATE event_log SET handled=1 WHERE id=?",
|
||||
(int(event_log_row_id),),
|
||||
)
|
||||
|
||||
data["_on_broadcast_committed"] = _on_commit
|
||||
data["_broadcast_audit"] = {"table": "fires", "pk": irwin_id}
|
||||
|
||||
|
||||
# ---------- helpers -------------------------------------------------------
|
||||
|
||||
|
||||
def _coerce_severity(sev: Any) -> Optional[str]:
|
||||
if sev is None: return None
|
||||
if isinstance(sev, str): return sev or None
|
||||
try: return str(int(sev))
|
||||
except (TypeError, ValueError): return str(sev)
|
||||
|
||||
|
||||
def _log_event(conn, *, now, source, category, severity_word, irwin_id,
|
||||
subject, handled, table_name, table_pk) -> None:
|
||||
"""Insert an event_log row; void return (used for tombstones/perimeters
|
||||
where the handled flag is fixed at write-time)."""
|
||||
conn.execute(
|
||||
"INSERT INTO event_log(received_at, source, category, severity_word, "
|
||||
"event_id_external, nats_subject, handled, table_name, table_pk) "
|
||||
"VALUES (?,?,?,?,?,?,?,?,?)",
|
||||
(now, source, category, severity_word, irwin_id, subject,
|
||||
int(bool(handled)), table_name, table_pk),
|
||||
)
|
||||
|
||||
|
||||
def _log_event_returning_id(conn, *, now, source, category, severity_word,
|
||||
irwin_id, subject, handled, table_name,
|
||||
table_pk) -> int:
|
||||
"""Insert an event_log row and return its primary key id.
|
||||
|
||||
Used for active-incident logging where the commit callback updates
|
||||
the same row to handled=1 once a broadcast actually goes out.
|
||||
"""
|
||||
cur = conn.execute(
|
||||
"INSERT INTO event_log(received_at, source, category, severity_word, "
|
||||
"event_id_external, nats_subject, handled, table_name, table_pk) "
|
||||
"VALUES (?,?,?,?,?,?,?,?,?)",
|
||||
(now, source, category, severity_word, irwin_id, subject,
|
||||
int(bool(handled)), table_name, table_pk),
|
||||
)
|
||||
return int(cur.lastrowid)
|
||||
|
||||
|
||||
# ---------- renderer ------------------------------------------------------
|
||||
|
||||
|
||||
def _render(n: dict, *, prefix: str = "") -> str:
|
||||
"""MEDIUM-style mesh wire string. See spec in module docstring."""
|
||||
name = n.get("incident_name") or "(unnamed)"
|
||||
itype = n.get("incident_type") or "incident"
|
||||
lat = n.get("lat")
|
||||
lon = n.get("lon")
|
||||
|
||||
anchor = _location_anchor(n)
|
||||
acres = n.get("acres")
|
||||
contained = n.get("contained_pct")
|
||||
|
||||
acres_str = "N/A" if acres is None else f"{int(acres):,} ac"
|
||||
contained_str = (
|
||||
"containment unknown" if contained is None
|
||||
else f"{int(contained)}% contained"
|
||||
)
|
||||
|
||||
prefix_str = f"{prefix}: " if prefix else ""
|
||||
head = f"🔥 {prefix_str}{name} ({itype}), {anchor}"
|
||||
body = f"{acres_str}, {contained_str}"
|
||||
|
||||
coords = ""
|
||||
if isinstance(lat, (int, float)) and isinstance(lon, (int, float)):
|
||||
coords = f", @ {lat:.3f},{lon:.3f}"
|
||||
|
||||
return f"{head}: {body}{coords}"
|
||||
|
||||
|
||||
def _location_anchor(n: dict) -> str:
|
||||
"""Anchor priority: geocoder.city > nearest_town > landclass > county."""
|
||||
city = n.get("geocoder_city")
|
||||
if city:
|
||||
return str(city)
|
||||
|
||||
lat = n.get("lat")
|
||||
lon = n.get("lon")
|
||||
if isinstance(lat, (int, float)) and isinstance(lon, (int, float)):
|
||||
try:
|
||||
from meshai.central_normalizer import nearest_town
|
||||
nt = nearest_town(lat, lon, max_distance_mi=100.0)
|
||||
except Exception:
|
||||
logger.exception("nearest_town failed; falling through")
|
||||
nt = None
|
||||
if nt and nt.get("name"):
|
||||
town = nt["name"]
|
||||
d = nt.get("distance_mi")
|
||||
bearing = nt.get("bearing")
|
||||
if isinstance(d, (int, float)):
|
||||
if d < 1:
|
||||
return f"near {town}"
|
||||
return f"{int(round(d))} mi {bearing or ''} of {town}".strip()
|
||||
return f"near {town}"
|
||||
|
||||
landclass = n.get("landclass")
|
||||
if landclass:
|
||||
return str(landclass)
|
||||
|
||||
county = n.get("county")
|
||||
state = n.get("state")
|
||||
if county and state:
|
||||
return f"{county} Co {state}"
|
||||
if state:
|
||||
return str(state)
|
||||
return "(location unknown)"
|
||||
|
|
@ -458,6 +458,288 @@ def _parse_state_511_atis(inner_data: dict, geo: dict) -> dict:
|
|||
}
|
||||
|
||||
|
||||
# ---------- wzdx federal vocabulary maps ----------------------------------
|
||||
|
||||
# FHWA WZDx v4 + custom-feed vocabulary observed in the wild. Unknown values
|
||||
# fall through to lowercased + hyphens→spaces (see _norm_wzdx_sub_type).
|
||||
_WZDX_WORK_TYPE_MAP: dict[str, Optional[str]] = {
|
||||
# WZDx v4 spec types_of_work.type_name enum:
|
||||
"maintenance": "maintenance",
|
||||
"minor-road-defect-repair": "minor repair",
|
||||
"roadside-work": "roadside work",
|
||||
"overhead-work": "overhead work",
|
||||
"below-road-work": "subsurface work",
|
||||
"barrier-work": "barrier work",
|
||||
"surface-work": "surface work",
|
||||
"painting": "painting",
|
||||
"roadway-relocation": "roadway relocation",
|
||||
"roadway-creation": "new construction",
|
||||
# Common informal values seen in upstream feeds (ID, WA):
|
||||
"road-work": "road work",
|
||||
"paving": "paving",
|
||||
"bridge-construction": "bridge construction",
|
||||
"bridge-maintenance": "bridge maintenance",
|
||||
"utility-work": "utility work",
|
||||
"road-construction": "road construction",
|
||||
"construction": "construction",
|
||||
"emergency-repairs": "emergency repairs",
|
||||
# event_type values (drop the too-generic ones):
|
||||
"work-zone": None,
|
||||
"detour": "detour",
|
||||
}
|
||||
|
||||
|
||||
# vehicle_impact taxonomy (WZDx v4). Maps to mesh-friendly phrase.
|
||||
# Returns None for values the renderer should drop entirely.
|
||||
_WZDX_IMPACT_MAP: dict[str, Optional[str]] = {
|
||||
"all-lanes-closed": "all lanes closed",
|
||||
"some-lanes-closed": "lanes reduced",
|
||||
"alternating-one-way": "one-way alternating",
|
||||
"unknown": None,
|
||||
"all-lanes-open": None, # informational only; nothing to do
|
||||
}
|
||||
|
||||
|
||||
def _norm_wzdx_sub_type(raw) -> Optional[str]:
|
||||
if not raw: return None
|
||||
s = str(raw).strip().lower()
|
||||
if not s: return None
|
||||
if s in _WZDX_WORK_TYPE_MAP:
|
||||
return _WZDX_WORK_TYPE_MAP[s]
|
||||
# Unknown value — keep lowercased, hyphens → spaces, single-line.
|
||||
return re.sub(r"\s+", " ", s.replace("-", " ")).strip() or None
|
||||
|
||||
|
||||
# ---------- per-adapter parser: wzdx federal ------------------------------
|
||||
|
||||
def _parse_wzdx_federal(inner_data: dict, geo: dict) -> dict:
|
||||
"""Normalize a wzdx-adapter envelope (FHWA WZDx federal spec).
|
||||
|
||||
Central flattens the upstream payload in practice (the FHWA-spec
|
||||
`core_details.*` nesting is not preserved), but we defensively check
|
||||
nested keys too so any future Central change doesn't silently regress.
|
||||
|
||||
sub_type uses types_of_work[0].type_name when present, else event_type,
|
||||
each normalized via _WZDX_WORK_TYPE_MAP. impact_phrase is folded INTO
|
||||
the sub_type slot for the renderer (so the description-slot reads e.g.
|
||||
'lanes reduced, paving' or 'one-way alternating' or 'road work').
|
||||
'all lanes closed' is set on impact='full_closure' so the renderer's
|
||||
existing full-closure promotion handles it -- avoids double-printing.
|
||||
"""
|
||||
cd = inner_data.get("core_details")
|
||||
if not isinstance(cd, dict): cd = {}
|
||||
def field(key):
|
||||
v = cd.get(key)
|
||||
if v is None or (isinstance(v, str) and not v.strip()):
|
||||
v = inner_data.get(key)
|
||||
return v
|
||||
|
||||
# --- road (raw, verbatim per Matt's spec) -----------------------------
|
||||
road_names = field("road_names")
|
||||
road = None
|
||||
if isinstance(road_names, list) and road_names:
|
||||
road = str(road_names[0]).strip() or None
|
||||
elif isinstance(road_names, str) and road_names.strip():
|
||||
road = road_names.strip()
|
||||
if _is_uninformative_road(road):
|
||||
road = None
|
||||
|
||||
# --- direction --------------------------------------------------------
|
||||
direction = _norm_direction(field("direction"))
|
||||
|
||||
# --- sub_type (types_of_work[0] | event_type) -------------------------
|
||||
work_type: Optional[str] = None
|
||||
tow = field("types_of_work")
|
||||
if isinstance(tow, list) and tow:
|
||||
first = tow[0]
|
||||
if isinstance(first, dict):
|
||||
work_type = _norm_wzdx_sub_type(first.get("type_name"))
|
||||
elif isinstance(first, str):
|
||||
work_type = _norm_wzdx_sub_type(first)
|
||||
if not work_type:
|
||||
work_type = _norm_wzdx_sub_type(field("event_type"))
|
||||
|
||||
# --- vehicle_impact ---------------------------------------------------
|
||||
vi_raw = (inner_data.get("vehicle_impact") or cd.get("vehicle_impact") or "")
|
||||
impact_phrase: Optional[str] = _WZDX_IMPACT_MAP.get(str(vi_raw).strip().lower())
|
||||
is_full_closure = (str(vi_raw).strip().lower() == "all-lanes-closed")
|
||||
|
||||
# Fold impact_phrase + work_type into the renderer's sub_type slot.
|
||||
# For full-closure, exclude impact_phrase here -- the renderer prepends
|
||||
# "all lanes closed" itself via the impact='full_closure' branch.
|
||||
parts: list[str] = []
|
||||
if impact_phrase and not is_full_closure:
|
||||
parts.append(impact_phrase)
|
||||
if work_type:
|
||||
parts.append(work_type)
|
||||
sub_type = ", ".join(parts) if parts else None
|
||||
impact = "full_closure" if is_full_closure else "partial"
|
||||
|
||||
# --- ends_at: structured end_date ISO-8601 ---------------------------
|
||||
ends_at: Optional[datetime] = None
|
||||
end_date = inner_data.get("end_date") or cd.get("end_date")
|
||||
if end_date:
|
||||
try:
|
||||
s = str(end_date).replace("Z", "+00:00")
|
||||
ends_at = datetime.fromisoformat(s)
|
||||
# Strip tzinfo so _format_end_short compares naive-to-naive.
|
||||
if ends_at.tzinfo is not None:
|
||||
ends_at = ends_at.astimezone().replace(tzinfo=None)
|
||||
except Exception:
|
||||
ends_at = None
|
||||
|
||||
# --- mile_start/_end: regex on description, fall back to structured --
|
||||
desc = _clean_description(field("description"))
|
||||
mile_start, mile_end = _parse_mile_posts(desc or "")
|
||||
if mile_start is None:
|
||||
ms = inner_data.get("road_mile_post_start")
|
||||
if ms is not None:
|
||||
try: mile_start = int(ms)
|
||||
except (TypeError, ValueError): pass
|
||||
if mile_end is None:
|
||||
me = inner_data.get("road_mile_post_end")
|
||||
if me is not None:
|
||||
try: mile_end = int(me)
|
||||
except (TypeError, ValueError): pass
|
||||
|
||||
# --- coordinates -----------------------------------------------------
|
||||
event_lat = inner_data.get("latitude")
|
||||
event_lon = inner_data.get("longitude")
|
||||
if event_lat is None and geo.get("centroid"):
|
||||
try: event_lon, event_lat = geo["centroid"][0], geo["centroid"][1]
|
||||
except (IndexError, TypeError): pass
|
||||
|
||||
# --- town fallback chain (same as state_511_atis) --------------------
|
||||
enriched = (inner_data.get("_enriched") or {}).get("geocoder") or {}
|
||||
town = (enriched.get("city") or "").strip() or None
|
||||
distance_mi: Optional[int] = None
|
||||
bearing: Optional[str] = None
|
||||
if town:
|
||||
distance_mi, bearing = _compute_distance_bearing(event_lat, event_lon, town)
|
||||
elif event_lat is not None:
|
||||
nt = nearest_town(event_lat, event_lon)
|
||||
if nt:
|
||||
town = nt.get("name")
|
||||
distance_mi = nt.get("distance_mi")
|
||||
bearing = nt.get("bearing")
|
||||
|
||||
return {
|
||||
"source": "wzdx",
|
||||
"road": road,
|
||||
"direction": direction,
|
||||
"mile_start": mile_start,
|
||||
"mile_end": mile_end,
|
||||
"description": desc,
|
||||
"sub_type": sub_type,
|
||||
"impact": impact,
|
||||
"ends_at": ends_at,
|
||||
"town": town,
|
||||
"distance_mi": distance_mi,
|
||||
"bearing": bearing,
|
||||
}
|
||||
|
||||
|
||||
|
||||
# ---------- WFIGS incidents (wildfire+prescribed) -------------------------
|
||||
|
||||
# IncidentName values like "IA 1", "IA 27" are auto-numbered Initial-Attack
|
||||
# placeholders that WFIGS issues before a fire gets a proper name. We pass
|
||||
# them through verbatim per Matt's call -- they at least signal "new fire
|
||||
# in <county>" even without an interesting name.
|
||||
_WFIGS_ACRES_KEYS = ("DailyAcres", "IncidentSize")
|
||||
_WFIGS_ACRES_RAW_KEYS = ("DiscoveryAcres", "FinalAcres")
|
||||
_WFIGS_CONTAINED_KEYS = ("PercentContained",)
|
||||
_WFIGS_CONTAINED_RAW_KEYS = ("PercentContained",)
|
||||
|
||||
|
||||
def _first_non_null(d: dict, keys) -> Any:
|
||||
"""Return d[k] for the first k in keys with a non-null value, else None."""
|
||||
for k in keys:
|
||||
v = d.get(k)
|
||||
if v is not None and v != "":
|
||||
return v
|
||||
return None
|
||||
|
||||
|
||||
def _parse_wfigs_acres(inner_data: dict) -> Optional[float]:
|
||||
"""Acres fallback chain: top-level DailyAcres/IncidentSize -> raw.* -> None."""
|
||||
val = _first_non_null(inner_data, _WFIGS_ACRES_KEYS)
|
||||
if val is None:
|
||||
raw = inner_data.get("raw") or {}
|
||||
if isinstance(raw, dict):
|
||||
val = _first_non_null(raw, _WFIGS_ACRES_RAW_KEYS)
|
||||
if val is None:
|
||||
return None
|
||||
try: return float(val)
|
||||
except (TypeError, ValueError): return None
|
||||
|
||||
|
||||
def _parse_wfigs_contained(inner_data: dict) -> Optional[int]:
|
||||
"""Containment fallback chain: top-level PercentContained -> raw.* -> None."""
|
||||
val = _first_non_null(inner_data, _WFIGS_CONTAINED_KEYS)
|
||||
if val is None:
|
||||
raw = inner_data.get("raw") or {}
|
||||
if isinstance(raw, dict):
|
||||
val = _first_non_null(raw, _WFIGS_CONTAINED_RAW_KEYS)
|
||||
if val is None:
|
||||
return None
|
||||
try: return int(round(float(val)))
|
||||
except (TypeError, ValueError): return None
|
||||
|
||||
|
||||
def _parse_wfigs_incidents(inner_data: dict, geo: dict) -> dict:
|
||||
"""Normalize a WFIGS-incidents payload into a flat render-ready dict.
|
||||
|
||||
Field shapes per Central v0.10.0 guide (see /OneDrive/.../wfigs-investigation.md):
|
||||
Top-level (incident): IrwinID, IncidentName, IncidentTypeCategory,
|
||||
latitude, longitude, FireDiscoveryDateTime (epoch-ms), POOState,
|
||||
POOCounty, DailyAcres, IncidentSize, PercentContained.
|
||||
Nested raw dict (97-key): DiscoveryAcres, FinalAcres, PercentContained
|
||||
(often the place where real values live in early season when the
|
||||
top-level fields haven't populated yet).
|
||||
_enriched.geocoder.landclass: optional ("Sawtooth National Forest", etc).
|
||||
|
||||
Returns the normalized dict. Caller layers on "_kind": "wfigs_incident".
|
||||
"""
|
||||
geocoder = geo.get("geocoder") or {}
|
||||
irwin_id = inner_data.get("IrwinID") or inner_data.get("irwin_id")
|
||||
name = inner_data.get("IncidentName")
|
||||
itype = inner_data.get("IncidentTypeCategory")
|
||||
lat = inner_data.get("latitude")
|
||||
lon = inner_data.get("longitude")
|
||||
county = inner_data.get("POOCounty")
|
||||
state = inner_data.get("POOState")
|
||||
landclass = geocoder.get("landclass")
|
||||
|
||||
# FireDiscoveryDateTime is epoch-ms in WFIGS; convert to epoch-s.
|
||||
declared_at_epoch = None
|
||||
fdt = inner_data.get("FireDiscoveryDateTime")
|
||||
if isinstance(fdt, (int, float)):
|
||||
# Heuristic: anything >1e12 is ms (post-2001 in ms is ~1.4e12).
|
||||
declared_at_epoch = int(fdt / 1000) if fdt > 1e12 else int(fdt)
|
||||
|
||||
acres = _parse_wfigs_acres(inner_data)
|
||||
contained_pct = _parse_wfigs_contained(inner_data)
|
||||
|
||||
# Geocoder-side anchor enrichment for the renderer.
|
||||
city = geocoder.get("city")
|
||||
|
||||
return {
|
||||
"irwin_id": irwin_id,
|
||||
"incident_name": name,
|
||||
"incident_type": itype,
|
||||
"acres": acres,
|
||||
"contained_pct": contained_pct,
|
||||
"lat": lat,
|
||||
"lon": lon,
|
||||
"county": county,
|
||||
"state": state,
|
||||
"landclass": landclass,
|
||||
"geocoder_city": city,
|
||||
"declared_at_epoch": declared_at_epoch,
|
||||
}
|
||||
|
||||
|
||||
# ---------- public entry point --------------------------------------------
|
||||
|
||||
def normalize(envelope: dict) -> Optional[dict]:
|
||||
|
|
@ -474,6 +756,34 @@ def normalize(envelope: dict) -> Optional[dict]:
|
|||
|
||||
if adapter == "state_511_atis":
|
||||
return _parse_state_511_atis(inner_data, geo)
|
||||
if adapter == "wzdx":
|
||||
return _parse_wzdx_federal(inner_data, geo)
|
||||
|
||||
# v0.5.8 WFIGS dispatch -- incidents + tombstones + perimeters.
|
||||
# The handler downstream uses _kind to route to change-detection
|
||||
# (active incidents) or to event_log-only logging (tombstones,
|
||||
# perimeters). Tombstones carry only irwin_id + state + county;
|
||||
# perimeters share the IrwinID with their parent incident.
|
||||
category_raw = inner.get("category") or ""
|
||||
if adapter == "wfigs_incidents":
|
||||
if category_raw.startswith("fire.incident.removed"):
|
||||
return {
|
||||
"_kind": "wfigs_tombstone",
|
||||
"irwin_id": inner_data.get("irwin_id") or inner_data.get("IrwinID"),
|
||||
"state": inner_data.get("state") or inner_data.get("POOState"),
|
||||
"county": inner_data.get("county") or inner_data.get("POOCounty"),
|
||||
}
|
||||
if category_raw.startswith("fire.incident"):
|
||||
n = _parse_wfigs_incidents(inner_data, geo)
|
||||
n["_kind"] = "wfigs_incident"
|
||||
return n
|
||||
if adapter == "wfigs_perimeters":
|
||||
return {
|
||||
"_kind": "wfigs_perimeter",
|
||||
"irwin_id": inner_data.get("irwin_id") or inner_data.get("IrwinID"),
|
||||
"state": inner_data.get("state") or inner_data.get("POOState"),
|
||||
"county": inner_data.get("county") or inner_data.get("POOCounty"),
|
||||
}
|
||||
|
||||
# Other adapters await per-adapter parsers; return None to defer.
|
||||
return None
|
||||
|
|
|
|||
|
|
@ -607,6 +607,13 @@ class NotificationsConfig:
|
|||
quiet_hours_enabled: bool = True # Master toggle for quiet hours
|
||||
quiet_hours_start: str = "22:00"
|
||||
quiet_hours_end: str = "06:00"
|
||||
# v0.5.8b cold-start grace: after the first event the dispatcher sees,
|
||||
# suppress mesh broadcasts for N seconds to absorb any JetStream
|
||||
# backlog. Persistence rows still get written -- only broadcasts are
|
||||
# suppressed. Anchor is "first-event-seen" (not container-boot) so
|
||||
# meshai can sit idle for hours with master OFF and the grace only
|
||||
# kicks in when adapters actually start producing.
|
||||
cold_start_grace_seconds: int = 60
|
||||
toggles: dict = field(default_factory=_default_toggles) # family -> NotificationToggle
|
||||
digest: DigestConfig = field(default_factory=DigestConfig)
|
||||
rules: list = field(default_factory=list) # List of NotificationRuleConfig
|
||||
|
|
|
|||
|
|
@ -51,6 +51,11 @@ class Dispatcher:
|
|||
self._stale_dropped = 0
|
||||
self._cooldown_dropped = 0
|
||||
self._dedup_dropped = 0
|
||||
# v0.5.8b cold-start grace: anchor lazily on FIRST event the
|
||||
# dispatcher sees through an enabled toggle. Grace window read
|
||||
# from config so it can be tuned at runtime via /api/config PUT.
|
||||
self._first_event_at: Optional[float] = None
|
||||
self._cold_start_dropped = 0
|
||||
# (toggle.name, category, region) -> last-fire wall-clock seconds
|
||||
self._toggle_cooldown: dict[tuple[str, str, str], float] = {}
|
||||
# Insertion-ordered (source, event.id) -> sentinel; evict oldest at cap.
|
||||
|
|
@ -115,6 +120,31 @@ class Dispatcher:
|
|||
if tog is None or not getattr(tog, "enabled", False):
|
||||
return
|
||||
|
||||
# ---------- Section 0 — cold-start grace (v0.5.8b) ----------
|
||||
# First event ever to reach an enabled toggle anchors the grace
|
||||
# window. Any broadcast attempt inside the window is dropped, but
|
||||
# the event still flowed through the consumer -> handler chain
|
||||
# before us, so persistence rows have already been written. Only
|
||||
# the broadcast is suppressed.
|
||||
grace_s = int(getattr(self._config.notifications, "cold_start_grace_seconds", 60) or 0)
|
||||
if grace_s > 0:
|
||||
now_anchor = time.time()
|
||||
if self._first_event_at is None:
|
||||
self._first_event_at = now_anchor
|
||||
self._logger.info(
|
||||
"cold-start grace anchor set: t0=%.3f window=%ds",
|
||||
now_anchor, grace_s,
|
||||
)
|
||||
if (now_anchor - self._first_event_at) < grace_s:
|
||||
self._cold_start_dropped += 1
|
||||
self._logger.info(
|
||||
"cold-start grace: dropping broadcast source=%s category=%s "
|
||||
"elapsed=%.1fs window=%ds",
|
||||
event.source, event.category,
|
||||
now_anchor - self._first_event_at, grace_s,
|
||||
)
|
||||
return
|
||||
|
||||
# ---------- Section 1 — staleness filter ----------
|
||||
# `event.timestamp` is the upstream-published wall-clock the adapter
|
||||
# sets when minting the event. For Central-sourced events that's the
|
||||
|
|
@ -200,6 +230,11 @@ class Dispatcher:
|
|||
success = await channel.deliver(payload, rule)
|
||||
if success:
|
||||
self._logger.info(f"Dispatched event {event.id} via toggle {fam}/{ch_type}")
|
||||
# v0.5.8b post-broadcast commit. Persistence-side
|
||||
# bookkeeping that should only happen when a delivery
|
||||
# actually went out: mesh_broadcasts_out audit row +
|
||||
# handler-supplied last_broadcast_* UPDATE callback.
|
||||
self._post_broadcast_commit(event, payload, rule, ch_type)
|
||||
else:
|
||||
self._logger.warning(f"Toggle channel delivery returned False for {fam}/{ch_type}")
|
||||
except Exception:
|
||||
|
|
@ -211,10 +246,66 @@ class Dispatcher:
|
|||
"stale_dropped": self._stale_dropped,
|
||||
"cooldown_dropped": self._cooldown_dropped,
|
||||
"dedup_dropped": self._dedup_dropped,
|
||||
"cold_start_dropped": self._cold_start_dropped,
|
||||
"cold_start_anchor_at": self._first_event_at,
|
||||
"cooldown_keys": len(self._toggle_cooldown),
|
||||
"dedup_lru_size": len(self._dedup_lru),
|
||||
}
|
||||
|
||||
def _post_broadcast_commit(self, event, payload, rule, ch_type: str) -> None:
|
||||
"""Persistence side-effects of an actually-successful broadcast.
|
||||
|
||||
Inserts the mesh_broadcasts_out audit row when the handler signalled
|
||||
it wants one via `event.data["_broadcast_audit"]`, then invokes the
|
||||
handler-supplied `_on_broadcast_committed` callback so the handler
|
||||
can refresh its own last_broadcast_* bookkeeping. Both calls are
|
||||
wrapped: a bookkeeping failure must NOT undo the actual broadcast
|
||||
nor break dispatch for sibling toggles.
|
||||
"""
|
||||
data = getattr(event, "data", None) or {}
|
||||
if not data:
|
||||
return
|
||||
committed_at = time.time()
|
||||
|
||||
audit = data.get("_broadcast_audit")
|
||||
if isinstance(audit, dict):
|
||||
try:
|
||||
from meshai.persistence import get_db
|
||||
conn = get_db()
|
||||
text = payload.message if payload is not None else (event.title or "")
|
||||
bytes_sent = len(text.encode("utf-8")) if text else 0
|
||||
if ch_type == "mesh_dm":
|
||||
node_ids = list(getattr(rule, "node_ids", []) or [])
|
||||
recipient = ",".join(map(str, node_ids)) or "dm"
|
||||
else:
|
||||
recipient = "broadcast"
|
||||
channel = getattr(rule, "broadcast_channel", None)
|
||||
conn.execute(
|
||||
"INSERT INTO mesh_broadcasts_out(sent_at, recipient, channel, "
|
||||
"text, source_event_table, source_event_pk, bytes_sent, "
|
||||
"ack_received) VALUES (?,?,?,?,?,?,?,?)",
|
||||
(
|
||||
int(committed_at), recipient, channel, text,
|
||||
audit.get("table"), audit.get("pk"),
|
||||
bytes_sent, 0,
|
||||
),
|
||||
)
|
||||
except Exception:
|
||||
self._logger.exception(
|
||||
"post-broadcast: mesh_broadcasts_out insert failed "
|
||||
"(table=%s pk=%s)",
|
||||
audit.get("table"), audit.get("pk"),
|
||||
)
|
||||
|
||||
cb = data.get("_on_broadcast_committed")
|
||||
if callable(cb):
|
||||
try:
|
||||
cb(committed_at)
|
||||
except Exception:
|
||||
self._logger.exception(
|
||||
"post-broadcast: handler commit-callback raised"
|
||||
)
|
||||
|
||||
def _toggle_to_rule(self, tog, ch_type: str, event: Event):
|
||||
from meshai.config import NotificationRuleConfig
|
||||
return NotificationRuleConfig(
|
||||
|
|
|
|||
31
meshai/persistence/__init__.py
Normal file
31
meshai/persistence/__init__.py
Normal file
|
|
@ -0,0 +1,31 @@
|
|||
"""meshai persistence layer.
|
||||
|
||||
SQLite-backed adapter event store + mesh telemetry. Single-writer pattern
|
||||
with WAL mode; threading.local connection pool. v1 schema covers 13 data
|
||||
tables + event_log + schema_meta.
|
||||
|
||||
Public API:
|
||||
from meshai.persistence import get_db, init_db, MESHAI_DB_PATH
|
||||
|
||||
The db.py module handles connection pooling and runs pending migrations
|
||||
on first init_db() call. Adapter handlers are responsible for inserting
|
||||
into the right table (none wired yet -- foundation only).
|
||||
"""
|
||||
|
||||
from meshai.persistence.db import (
|
||||
get_db,
|
||||
init_db,
|
||||
close_thread_connection,
|
||||
MESHAI_DB_PATH,
|
||||
DEFAULT_DB_PATH,
|
||||
SCHEMA_VERSION,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"get_db",
|
||||
"init_db",
|
||||
"close_thread_connection",
|
||||
"MESHAI_DB_PATH",
|
||||
"DEFAULT_DB_PATH",
|
||||
"SCHEMA_VERSION",
|
||||
]
|
||||
194
meshai/persistence/db.py
Normal file
194
meshai/persistence/db.py
Normal file
|
|
@ -0,0 +1,194 @@
|
|||
"""SQLite persistence connection management + migration runner.
|
||||
|
||||
Single-writer SQLite pattern with WAL journal mode for reader concurrency.
|
||||
One connection per thread (threading.local) -- callers should not share
|
||||
connections across threads.
|
||||
|
||||
Path resolution:
|
||||
1. MESHAI_DB_PATH env var (explicit override)
|
||||
2. DEFAULT_DB_PATH = /data/meshai.sqlite (prod container mount)
|
||||
|
||||
Special value ":memory:" or any path containing "memory" routes to an
|
||||
in-memory SQLite for tests.
|
||||
|
||||
Migrations live in meshai/persistence/migrations/v*.sql. The runner
|
||||
applies them in version order, recording the applied version in
|
||||
schema_meta. Idempotent re-run is a no-op.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
import sqlite3
|
||||
import threading
|
||||
from pathlib import Path
|
||||
from typing import Iterable, Optional
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
DEFAULT_DB_PATH = "/data/meshai.sqlite"
|
||||
MESHAI_DB_PATH_ENV = "MESHAI_DB_PATH"
|
||||
SCHEMA_VERSION = 1
|
||||
SCHEMA_META_TABLE = "schema_meta"
|
||||
MIGRATIONS_DIR = Path(__file__).parent / "migrations"
|
||||
|
||||
# Per-thread connection pool. Each thread that calls get_db() gets its
|
||||
# own sqlite3.Connection cached on threading.local. Tests can clear
|
||||
# via close_thread_connection() between cases.
|
||||
_local = threading.local()
|
||||
# Module-level lock guards init_db() so concurrent first-callers don't
|
||||
# race on migration application.
|
||||
_init_lock = threading.Lock()
|
||||
# Cache of initialised database paths in this process so init_db() is
|
||||
# idempotent without re-reading migration files on every call.
|
||||
_initialised: set[str] = set()
|
||||
|
||||
|
||||
def MESHAI_DB_PATH() -> str:
|
||||
"""Resolve the active SQLite path (env var override or default)."""
|
||||
return os.environ.get(MESHAI_DB_PATH_ENV) or DEFAULT_DB_PATH
|
||||
|
||||
|
||||
def _is_memory_path(path: str) -> bool:
|
||||
return path == ":memory:" or "mode=memory" in path or path.startswith("file::memory:")
|
||||
|
||||
|
||||
def _connect(path: str) -> sqlite3.Connection:
|
||||
"""Open a SQLite connection with sane defaults for this project."""
|
||||
if _is_memory_path(path):
|
||||
# For in-memory tests, use a shared cache so multiple connections
|
||||
# in the same thread can see the same DB. Tests that want isolation
|
||||
# call close_thread_connection() between cases.
|
||||
uri = "file::memory:?cache=shared"
|
||||
conn = sqlite3.connect(uri, uri=True, isolation_level=None,
|
||||
check_same_thread=False)
|
||||
else:
|
||||
# Ensure parent dir exists for file-backed DBs.
|
||||
Path(path).parent.mkdir(parents=True, exist_ok=True)
|
||||
conn = sqlite3.connect(path, isolation_level=None,
|
||||
check_same_thread=False, timeout=30.0)
|
||||
conn.row_factory = sqlite3.Row
|
||||
# Enable foreign keys (off by default in SQLite); WAL mode for reader
|
||||
# concurrency; reasonable busy timeout for the single-writer pattern.
|
||||
conn.execute("PRAGMA foreign_keys = ON")
|
||||
if not _is_memory_path(path):
|
||||
conn.execute("PRAGMA journal_mode = WAL")
|
||||
conn.execute("PRAGMA synchronous = NORMAL")
|
||||
conn.execute("PRAGMA busy_timeout = 30000")
|
||||
return conn
|
||||
|
||||
|
||||
def get_db(path: Optional[str] = None) -> sqlite3.Connection:
|
||||
"""Return a SQLite connection for the current thread (cached).
|
||||
|
||||
First call initialises the database (runs pending migrations).
|
||||
Subsequent calls in the same thread return the cached connection.
|
||||
"""
|
||||
target = path or MESHAI_DB_PATH()
|
||||
cached = getattr(_local, "conn", None)
|
||||
cached_path = getattr(_local, "path", None)
|
||||
if cached is not None and cached_path == target:
|
||||
return cached
|
||||
# Different path requested or no cached conn -- (re)open.
|
||||
if cached is not None:
|
||||
try: cached.close()
|
||||
except Exception: pass
|
||||
conn = _connect(target)
|
||||
_local.conn = conn
|
||||
_local.path = target
|
||||
if target not in _initialised:
|
||||
with _init_lock:
|
||||
if target not in _initialised:
|
||||
_apply_migrations(conn)
|
||||
_initialised.add(target)
|
||||
return conn
|
||||
|
||||
|
||||
def close_thread_connection() -> None:
|
||||
"""Close + drop the cached connection for the current thread.
|
||||
|
||||
Tests call this between cases to ensure a clean slate. The shared-cache
|
||||
in-memory database is reset on the LAST close in the process.
|
||||
"""
|
||||
conn = getattr(_local, "conn", None)
|
||||
if conn is not None:
|
||||
try: conn.close()
|
||||
except Exception: pass
|
||||
if hasattr(_local, "conn"): del _local.conn
|
||||
if hasattr(_local, "path"): del _local.path
|
||||
|
||||
|
||||
def init_db(path: Optional[str] = None) -> sqlite3.Connection:
|
||||
"""Explicit init entry point (idempotent). Equivalent to get_db()
|
||||
semantically but documents intent at startup. Returns the connection."""
|
||||
return get_db(path)
|
||||
|
||||
|
||||
def _read_migration_files() -> list[tuple[int, str, str]]:
|
||||
"""Return [(version_int, filename, sql_text), ...] sorted ascending."""
|
||||
if not MIGRATIONS_DIR.is_dir():
|
||||
return []
|
||||
out: list[tuple[int, str, str]] = []
|
||||
for p in sorted(MIGRATIONS_DIR.iterdir()):
|
||||
if not p.is_file() or p.suffix.lower() != ".sql":
|
||||
continue
|
||||
# Filename format: v<N>.sql or v<N>_<label>.sql
|
||||
stem = p.stem
|
||||
if not stem.startswith("v"):
|
||||
continue
|
||||
n_str = stem[1:].split("_", 1)[0]
|
||||
try: n = int(n_str)
|
||||
except ValueError: continue
|
||||
out.append((n, p.name, p.read_text()))
|
||||
return out
|
||||
|
||||
|
||||
def _current_version(conn: sqlite3.Connection) -> int:
|
||||
"""Return the highest applied migration version, or 0 if none."""
|
||||
# Does the schema_meta table exist?
|
||||
row = conn.execute(
|
||||
"SELECT name FROM sqlite_master WHERE type='table' AND name=?",
|
||||
(SCHEMA_META_TABLE,),
|
||||
).fetchone()
|
||||
if row is None:
|
||||
return 0
|
||||
row = conn.execute(
|
||||
f"SELECT value FROM {SCHEMA_META_TABLE} WHERE key='version'",
|
||||
).fetchone()
|
||||
if row is None:
|
||||
return 0
|
||||
try: return int(row["value"])
|
||||
except (TypeError, ValueError): return 0
|
||||
|
||||
|
||||
def _apply_migrations(conn: sqlite3.Connection) -> None:
|
||||
"""Apply any pending migrations in version order. Idempotent."""
|
||||
migrations = _read_migration_files()
|
||||
if not migrations:
|
||||
logger.warning("persistence: no migration files found in %s", MIGRATIONS_DIR)
|
||||
return
|
||||
current = _current_version(conn)
|
||||
pending = [(n, name, sql) for n, name, sql in migrations if n > current]
|
||||
if not pending:
|
||||
logger.debug("persistence: schema up to date at v%d", current)
|
||||
return
|
||||
for version, filename, sql in pending:
|
||||
logger.info("persistence: applying migration %s (v%d)", filename, version)
|
||||
# sqlite3.Connection.executescript() ISSUES ITS OWN COMMIT before
|
||||
# starting the script and another at the end, so wrapping it in an
|
||||
# explicit BEGIN/COMMIT is both unnecessary and broken (the
|
||||
# ROLLBACK in the except clause would fire against an already-
|
||||
# committed-or-empty transaction). Migration atomicity is bounded
|
||||
# by executescript's own transaction.
|
||||
try:
|
||||
conn.executescript(sql)
|
||||
conn.execute(
|
||||
f"INSERT OR REPLACE INTO {SCHEMA_META_TABLE}(key, value) VALUES('version', ?)",
|
||||
(str(version),),
|
||||
)
|
||||
except Exception:
|
||||
logger.exception("persistence: migration %s failed", filename)
|
||||
raise
|
||||
logger.info("persistence: schema now at v%d", pending[-1][0])
|
||||
289
meshai/persistence/migrations/v1.sql
Normal file
289
meshai/persistence/migrations/v1.sql
Normal file
|
|
@ -0,0 +1,289 @@
|
|||
-- v0.5.8 persistence foundation: 13 data tables + event_log + schema_meta.
|
||||
-- All timestamps are INTEGER epoch seconds (UTC) unless otherwise noted.
|
||||
-- Booleans are INTEGER 0/1.
|
||||
|
||||
-- ============================================================================
|
||||
-- schema_meta: key/value store; version key tracks applied migration.
|
||||
-- ============================================================================
|
||||
CREATE TABLE IF NOT EXISTS schema_meta (
|
||||
key TEXT PRIMARY KEY,
|
||||
value TEXT NOT NULL
|
||||
);
|
||||
|
||||
-- ============================================================================
|
||||
-- CROSS-CUTTING: event_log
|
||||
-- Every envelope received by the consumer lands here for auditing. Adapter
|
||||
-- handlers fill table_name / table_pk after inserting into their specific
|
||||
-- table; handled=1 marks success.
|
||||
-- ============================================================================
|
||||
CREATE TABLE IF NOT EXISTS event_log (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
received_at INTEGER NOT NULL,
|
||||
source TEXT NOT NULL,
|
||||
category TEXT,
|
||||
severity_word TEXT,
|
||||
event_id_external TEXT,
|
||||
nats_subject TEXT,
|
||||
handled INTEGER NOT NULL DEFAULT 0,
|
||||
table_name TEXT,
|
||||
table_pk TEXT
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_event_log_received ON event_log(received_at);
|
||||
CREATE INDEX IF NOT EXISTS idx_event_log_source ON event_log(source, received_at);
|
||||
CREATE INDEX IF NOT EXISTS idx_event_log_handled ON event_log(handled, received_at);
|
||||
|
||||
-- ============================================================================
|
||||
-- traffic_events (state_511_atis + wzdx_federal)
|
||||
-- Composite PK (source, external_id) so the same envelope re-arriving on
|
||||
-- restart doesn't duplicate. last_seen_at gets bumped on every reissue;
|
||||
-- last_broadcast_at records the most recent mesh broadcast for cooldown.
|
||||
-- ============================================================================
|
||||
CREATE TABLE IF NOT EXISTS traffic_events (
|
||||
source TEXT NOT NULL, -- 'state_511_atis' | 'wzdx'
|
||||
external_id TEXT NOT NULL,
|
||||
road TEXT,
|
||||
direction TEXT,
|
||||
mile_start INTEGER,
|
||||
mile_end INTEGER,
|
||||
county TEXT,
|
||||
state TEXT,
|
||||
lat REAL,
|
||||
lon REAL,
|
||||
sub_type TEXT,
|
||||
impact TEXT,
|
||||
start_at INTEGER,
|
||||
end_at INTEGER,
|
||||
first_seen_at INTEGER NOT NULL,
|
||||
last_seen_at INTEGER NOT NULL,
|
||||
last_broadcast_at INTEGER,
|
||||
PRIMARY KEY (source, external_id)
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_traffic_last_seen ON traffic_events(last_seen_at);
|
||||
CREATE INDEX IF NOT EXISTS idx_traffic_last_broadcast ON traffic_events(last_broadcast_at);
|
||||
CREATE INDEX IF NOT EXISTS idx_traffic_state_county ON traffic_events(state, county);
|
||||
|
||||
-- ============================================================================
|
||||
-- fires (wfigs_incidents)
|
||||
-- PK is IRWIN UUID. last_broadcast_acres + last_broadcast_contained enable
|
||||
-- the 8h-rate-limit change-detection so subsequent updates that don't move
|
||||
-- acres or containment can be suppressed.
|
||||
-- ============================================================================
|
||||
CREATE TABLE IF NOT EXISTS fires (
|
||||
irwin_id TEXT PRIMARY KEY,
|
||||
incident_name TEXT,
|
||||
incident_type TEXT,
|
||||
current_acres REAL,
|
||||
current_contained_pct INTEGER,
|
||||
status TEXT,
|
||||
lat REAL,
|
||||
lon REAL,
|
||||
county TEXT,
|
||||
state TEXT,
|
||||
landclass TEXT,
|
||||
declared_at INTEGER,
|
||||
last_event_at INTEGER NOT NULL,
|
||||
last_broadcast_at INTEGER,
|
||||
last_broadcast_acres REAL,
|
||||
last_broadcast_contained INTEGER
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_fires_last_event ON fires(last_event_at);
|
||||
CREATE INDEX IF NOT EXISTS idx_fires_last_broadcast ON fires(last_broadcast_at);
|
||||
CREATE INDEX IF NOT EXISTS idx_fires_state ON fires(state);
|
||||
|
||||
-- ============================================================================
|
||||
-- firms_pixels (v0.6 use; schema-ready now per Matt's request)
|
||||
-- Nullable FK to fires(irwin_id) so unattached hotspots store without join.
|
||||
-- v0.6 fire-tracker will populate irwin_id via spatial join.
|
||||
-- ============================================================================
|
||||
CREATE TABLE IF NOT EXISTS firms_pixels (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
irwin_id TEXT REFERENCES fires(irwin_id) ON DELETE SET NULL,
|
||||
lat REAL NOT NULL,
|
||||
lon REAL NOT NULL,
|
||||
acq_time INTEGER NOT NULL,
|
||||
frp REAL,
|
||||
confidence TEXT,
|
||||
satellite TEXT,
|
||||
brightness REAL
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_firms_pixels_acq_time ON firms_pixels(acq_time);
|
||||
CREATE INDEX IF NOT EXISTS idx_firms_pixels_irwin ON firms_pixels(irwin_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_firms_pixels_latlon ON firms_pixels(lat, lon);
|
||||
|
||||
-- ============================================================================
|
||||
-- quake_events (usgs_quake)
|
||||
-- PK is USGS event id (e.g. ci10240102).
|
||||
-- ============================================================================
|
||||
CREATE TABLE IF NOT EXISTS quake_events (
|
||||
event_id TEXT PRIMARY KEY,
|
||||
magnitude REAL,
|
||||
depth_km REAL,
|
||||
place TEXT,
|
||||
lat REAL,
|
||||
lon REAL,
|
||||
occurred_at INTEGER,
|
||||
tsunami_warning INTEGER NOT NULL DEFAULT 0,
|
||||
first_seen_at INTEGER NOT NULL,
|
||||
last_broadcast_at INTEGER
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_quake_occurred ON quake_events(occurred_at);
|
||||
CREATE INDEX IF NOT EXISTS idx_quake_last_broadcast ON quake_events(last_broadcast_at);
|
||||
|
||||
-- ============================================================================
|
||||
-- nws_alerts (nws)
|
||||
-- PK is NWS CAP id (urn-style).
|
||||
-- ============================================================================
|
||||
CREATE TABLE IF NOT EXISTS nws_alerts (
|
||||
event_id TEXT PRIMARY KEY,
|
||||
alert_type TEXT,
|
||||
severity TEXT,
|
||||
county TEXT,
|
||||
state TEXT,
|
||||
headline TEXT,
|
||||
description TEXT,
|
||||
expires_at INTEGER,
|
||||
first_seen_at INTEGER NOT NULL,
|
||||
last_broadcast_at INTEGER
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_nws_expires ON nws_alerts(expires_at);
|
||||
CREATE INDEX IF NOT EXISTS idx_nws_state_severity ON nws_alerts(state, severity);
|
||||
|
||||
-- ============================================================================
|
||||
-- gauge_readings (usgs nwis) -- time-series, autoincrement PK.
|
||||
-- ============================================================================
|
||||
CREATE TABLE IF NOT EXISTS gauge_readings (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
site_id TEXT NOT NULL,
|
||||
gauge_name TEXT,
|
||||
reading_value REAL,
|
||||
reading_unit TEXT,
|
||||
threshold_state TEXT, -- normal | action | flood_minor | flood_moderate | flood_major
|
||||
flow_cfs REAL,
|
||||
reading_time INTEGER NOT NULL,
|
||||
lat REAL,
|
||||
lon REAL
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_gauge_site_time ON gauge_readings(site_id, reading_time);
|
||||
CREATE INDEX IF NOT EXISTS idx_gauge_time ON gauge_readings(reading_time);
|
||||
CREATE INDEX IF NOT EXISTS idx_gauge_threshold ON gauge_readings(threshold_state, reading_time);
|
||||
|
||||
-- ============================================================================
|
||||
-- swpc_events (swpc) -- covers kindex / proton / alert.
|
||||
-- payload_json holds the full structured payload for the event_type so we
|
||||
-- don't have to migrate the schema every time a new SWPC product surfaces.
|
||||
-- ============================================================================
|
||||
CREATE TABLE IF NOT EXISTS swpc_events (
|
||||
event_id TEXT PRIMARY KEY,
|
||||
event_type TEXT NOT NULL, -- kindex | proton | alert
|
||||
severity_int INTEGER,
|
||||
payload_json TEXT,
|
||||
occurred_at INTEGER NOT NULL,
|
||||
first_seen_at INTEGER NOT NULL,
|
||||
last_broadcast_at INTEGER
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_swpc_occurred ON swpc_events(occurred_at);
|
||||
CREATE INDEX IF NOT EXISTS idx_swpc_type_time ON swpc_events(event_type, occurred_at);
|
||||
|
||||
-- ============================================================================
|
||||
-- MESH TABLES
|
||||
-- ============================================================================
|
||||
|
||||
-- mesh_nodes: per-node latest-known state + first/last seen + stale flag.
|
||||
CREATE TABLE IF NOT EXISTS mesh_nodes (
|
||||
node_id TEXT PRIMARY KEY, -- '!aabbccdd' or num as str
|
||||
long_name TEXT,
|
||||
short_name TEXT,
|
||||
hw_model TEXT,
|
||||
last_lat REAL,
|
||||
last_lon REAL,
|
||||
last_altitude REAL,
|
||||
last_battery_pct INTEGER,
|
||||
last_voltage REAL,
|
||||
last_rssi INTEGER,
|
||||
last_snr REAL,
|
||||
neighbor_count INTEGER,
|
||||
first_seen_at INTEGER NOT NULL,
|
||||
last_seen_at INTEGER NOT NULL,
|
||||
is_stale INTEGER NOT NULL DEFAULT 0
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_mesh_nodes_last_seen ON mesh_nodes(last_seen_at);
|
||||
CREATE INDEX IF NOT EXISTS idx_mesh_nodes_is_stale ON mesh_nodes(is_stale);
|
||||
|
||||
-- mesh_telemetry: time-series; bucket='raw' incoming, rolled up to 'hourly'
|
||||
-- / 'daily' by a retention job (not part of foundation).
|
||||
CREATE TABLE IF NOT EXISTS mesh_telemetry (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
node_id TEXT NOT NULL,
|
||||
recorded_at INTEGER NOT NULL,
|
||||
battery_pct INTEGER,
|
||||
voltage REAL,
|
||||
current_ma REAL,
|
||||
channel_util REAL,
|
||||
air_util_tx REAL,
|
||||
uptime_seconds INTEGER,
|
||||
bucket TEXT NOT NULL DEFAULT 'raw'
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_mesh_tel_node_time ON mesh_telemetry(node_id, recorded_at);
|
||||
CREATE INDEX IF NOT EXISTS idx_mesh_tel_time ON mesh_telemetry(recorded_at);
|
||||
CREATE INDEX IF NOT EXISTS idx_mesh_tel_bucket ON mesh_telemetry(bucket, recorded_at);
|
||||
|
||||
-- mesh_positions: time-series; bucket logic same as telemetry.
|
||||
CREATE TABLE IF NOT EXISTS mesh_positions (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
node_id TEXT NOT NULL,
|
||||
recorded_at INTEGER NOT NULL,
|
||||
lat REAL,
|
||||
lon REAL,
|
||||
altitude REAL,
|
||||
ground_speed REAL,
|
||||
bucket TEXT NOT NULL DEFAULT 'raw'
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_mesh_pos_node_time ON mesh_positions(node_id, recorded_at);
|
||||
CREATE INDEX IF NOT EXISTS idx_mesh_pos_time ON mesh_positions(recorded_at);
|
||||
CREATE INDEX IF NOT EXISTS idx_mesh_pos_bucket ON mesh_positions(bucket, recorded_at);
|
||||
|
||||
-- mesh_messages_in: inbound text traffic (broadcasts + DMs).
|
||||
CREATE TABLE IF NOT EXISTS mesh_messages_in (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
from_node TEXT NOT NULL,
|
||||
to_node TEXT, -- NULL = channel broadcast
|
||||
channel INTEGER,
|
||||
text TEXT,
|
||||
received_at INTEGER NOT NULL,
|
||||
portnum TEXT,
|
||||
payload_size INTEGER
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_mesh_msg_in_received ON mesh_messages_in(received_at);
|
||||
CREATE INDEX IF NOT EXISTS idx_mesh_msg_in_from_node ON mesh_messages_in(from_node, received_at);
|
||||
|
||||
-- mesh_broadcasts_out: outbound meshai-originated mesh deliveries.
|
||||
-- source_event_table/source_event_pk back-links to the table+pk that
|
||||
-- triggered the broadcast (e.g. 'fires' / IRWIN id).
|
||||
CREATE TABLE IF NOT EXISTS mesh_broadcasts_out (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
sent_at INTEGER NOT NULL,
|
||||
recipient TEXT NOT NULL, -- 'broadcast' or node_id
|
||||
channel INTEGER,
|
||||
text TEXT,
|
||||
source_event_table TEXT,
|
||||
source_event_pk TEXT,
|
||||
bytes_sent INTEGER,
|
||||
ack_received INTEGER NOT NULL DEFAULT 0
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_mesh_bcast_sent ON mesh_broadcasts_out(sent_at);
|
||||
CREATE INDEX IF NOT EXISTS idx_mesh_bcast_source ON mesh_broadcasts_out(source_event_table, source_event_pk);
|
||||
|
||||
-- mesh_health_events: derived signal events (node went stale, recovered,
|
||||
-- routing change detected, mesh score dropped, etc.). detail_json is the
|
||||
-- structured payload; event_type is the canonical short tag.
|
||||
CREATE TABLE IF NOT EXISTS mesh_health_events (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
event_type TEXT NOT NULL,
|
||||
node_id TEXT,
|
||||
detected_at INTEGER NOT NULL,
|
||||
severity TEXT,
|
||||
detail_json TEXT
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_mesh_hevt_detected ON mesh_health_events(detected_at);
|
||||
CREATE INDEX IF NOT EXISTS idx_mesh_hevt_type ON mesh_health_events(event_type, detected_at);
|
||||
CREATE INDEX IF NOT EXISTS idx_mesh_hevt_node ON mesh_health_events(node_id, detected_at);
|
||||
|
|
@ -392,3 +392,284 @@ def test_geocoder_name_is_never_used_as_town_fallback(monkeypatch):
|
|||
n = normalize(env)
|
||||
# Must NOT pick up "Cache Nf Road 444" from geocoder.name.
|
||||
assert n["town"] is None
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# v0.5.8-wzdx federal parser tests
|
||||
# ============================================================================
|
||||
|
||||
# --- representative envelopes (flat shape, as Central actually publishes) ---
|
||||
|
||||
_WZDX_ID_FULL = {
|
||||
"data": {
|
||||
"adapter": "wzdx",
|
||||
"category": "work_zone.wzdx",
|
||||
"time": "2026-06-01T13:00:00Z",
|
||||
"severity": 3,
|
||||
"geo": {"centroid": [-112.408309608311, 43.0208066348276],
|
||||
"primary_region": "US-ID", "regions": ["US-ID"]},
|
||||
"data": {
|
||||
"road_names": ["Exit 80 On Ramp"],
|
||||
"direction": "southbound",
|
||||
"description": " Road construction on Exit 80 On Ramp Southbound near MM (80)."
|
||||
" All lanes closed. 6/1/2026 7:00 AM to 6/10/2026 6:00 PM Mon, Tue ...",
|
||||
"vehicle_impact": "all-lanes-closed",
|
||||
"event_status": None,
|
||||
"start_date": "2026-06-01T13:00:00Z",
|
||||
"end_date": "2026-06-11T00:00:00Z",
|
||||
"data_source_id": "ERS",
|
||||
"feed_name": "iddot",
|
||||
"feed_state_code": "ID",
|
||||
"latitude": 43.0208066348276,
|
||||
"longitude": -112.408309608311,
|
||||
"_enriched": {"geocoder": {"city": None, "name": "Ross Fork Creek",
|
||||
"county": "Bannock", "state": "Idaho"}},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
_WZDX_WA = {
|
||||
"data": {
|
||||
"adapter": "wzdx",
|
||||
"category": "work_zone.wzdx",
|
||||
"time": "2026-06-01T00:00:00+00:00",
|
||||
"severity": 1,
|
||||
"geo": {"centroid": [-117.33633, 46.433365], "primary_region": "US-WA"},
|
||||
"data": {
|
||||
"road_names": ["012"],
|
||||
"direction": "westbound",
|
||||
"description": "Contract - XE3608 SR 12",
|
||||
"vehicle_impact": "unknown",
|
||||
"event_status": "pending",
|
||||
"start_date": "2026-06-01T00:00:00+00:00",
|
||||
"end_date": "2026-06-05T00:00:00+00:00",
|
||||
"data_source_id": "WSDOT-WZDB",
|
||||
"feed_name": "wsdot",
|
||||
"feed_state_code": "WA",
|
||||
"latitude": 46.433365,
|
||||
"longitude": -117.33633,
|
||||
"_enriched": {"geocoder": {"city": None, "name": "US Highway 12",
|
||||
"county": "Garfield", "state": "Washington"}},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
_WZDX_MCCALL = {
|
||||
"data": {
|
||||
"adapter": "wzdx", "category": "work_zone.wzdx",
|
||||
"time": "2026-05-28T23:00:00Z", "severity": 1,
|
||||
"geo": {"centroid": [-116.09759, 44.9065083834611], "primary_region": "US-ID"},
|
||||
"data": {
|
||||
"road_names": ["SH-55"],
|
||||
"direction": "unknown",
|
||||
"description": " Emergency repairs on SH-55 Both Directions near Washington St."
|
||||
" 5/28/2026 5:00 PM to 5/29/2026 8:00 AM Thu, Fri: ...",
|
||||
"vehicle_impact": "all-lanes-open",
|
||||
"start_date": "2026-05-28T23:00:00Z",
|
||||
"end_date": "2026-05-29T14:00:00Z",
|
||||
"feed_state_code": "ID",
|
||||
"latitude": 44.9065083834611,
|
||||
"longitude": -116.09759,
|
||||
"_enriched": {"geocoder": {"city": "McCall", "county": "Valley", "state": "ID"}},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _normalize_wzdx(env):
|
||||
n = normalize(env)
|
||||
assert n is not None
|
||||
assert n["source"] == "wzdx"
|
||||
return n
|
||||
|
||||
|
||||
# --- (a) Idaho wzdx full-field parse ---------------------------------------
|
||||
|
||||
def test_wzdx_idaho_full_fields_normalized(monkeypatch):
|
||||
_clear_h3_cache()
|
||||
from meshai import central_normalizer as cn
|
||||
# Mock Photon for the SECONDARY town path (city is null in this envelope).
|
||||
monkeypatch.setattr(cn, "_photon_reverse_places",
|
||||
lambda lat, lon: [
|
||||
{"geometry": {"coordinates": [-112.4373, 43.0299]},
|
||||
"properties": {"name": "Fort Hall",
|
||||
"osm_key": "place", "osm_value": "village"}},
|
||||
])
|
||||
n = _normalize_wzdx(_WZDX_ID_FULL)
|
||||
assert n["road"] is None # Exit-ramp pattern → uninformative-road drop
|
||||
assert n["direction"] == "southbound"
|
||||
# sub_type combines impact-phrase (suppressed under full-closure) + work_type
|
||||
# (None here — types_of_work absent). With full-closure, sub_type stays None
|
||||
# and the renderer prepends "all lanes closed".
|
||||
assert n["sub_type"] is None
|
||||
assert n["impact"] == "full_closure"
|
||||
assert n["mile_start"] == 80 and n["mile_end"] is None
|
||||
assert n["town"] == "Fort Hall" # via Photon nearest_town
|
||||
assert isinstance(n["ends_at"], datetime)
|
||||
assert n["ends_at"].year == 2026 and n["ends_at"].month == 6 and n["ends_at"].day == 11
|
||||
|
||||
|
||||
def test_wzdx_wa_road_passes_through_verbatim(monkeypatch):
|
||||
_clear_h3_cache()
|
||||
from meshai import central_normalizer as cn
|
||||
monkeypatch.setattr(cn, "_photon_reverse_places", lambda lat, lon: [])
|
||||
n = _normalize_wzdx(_WZDX_WA)
|
||||
# Per spec: "honor upstream verbatim, no expansion" -- raw '012' passes through.
|
||||
assert n["road"] == "012"
|
||||
assert n["direction"] == "westbound"
|
||||
# vehicle_impact='unknown' → impact_phrase=None; sub_type stays None.
|
||||
assert n["sub_type"] is None
|
||||
assert n["impact"] == "partial"
|
||||
# No MM in WA descriptions; mile_start stays None.
|
||||
assert n["mile_start"] is None
|
||||
assert isinstance(n["ends_at"], datetime)
|
||||
assert n["town"] is None # city null + Photon returned no places
|
||||
|
||||
|
||||
# --- (c) vehicle_impact mapping for each main value ------------------------
|
||||
|
||||
@pytest.mark.parametrize("vi_raw,expected_sub_type,expected_impact", [
|
||||
("all-lanes-closed", None, "full_closure"),
|
||||
("some-lanes-closed", "lanes reduced", "partial"),
|
||||
("alternating-one-way", "one-way alternating", "partial"),
|
||||
("unknown", None, "partial"),
|
||||
("all-lanes-open", None, "partial"),
|
||||
("totally-made-up", None, "partial"),
|
||||
])
|
||||
def test_wzdx_vehicle_impact_mapping(vi_raw, expected_sub_type, expected_impact, monkeypatch):
|
||||
_clear_h3_cache()
|
||||
from meshai import central_normalizer as cn
|
||||
monkeypatch.setattr(cn, "_photon_reverse_places", lambda lat, lon: [])
|
||||
env = {"data": {"adapter": "wzdx", "category": "work_zone.wzdx", "time": "2026-06-01T00:00:00Z",
|
||||
"geo": {"centroid": [-116.0, 44.0]},
|
||||
"data": {"road_names": ["SH-1"], "direction": "northbound",
|
||||
"description": "X", "vehicle_impact": vi_raw,
|
||||
"end_date": "2026-06-05T17:00:00Z",
|
||||
"latitude": 44.0, "longitude": -116.0,
|
||||
"_enriched": {"geocoder": {"city": "Boise"}}}}}
|
||||
n = normalize(env)
|
||||
assert n["sub_type"] == expected_sub_type
|
||||
assert n["impact"] == expected_impact
|
||||
|
||||
|
||||
# --- (d) structured end_date parses to friendly format --------------------
|
||||
|
||||
def test_wzdx_end_date_iso_parsed_to_datetime(monkeypatch):
|
||||
_clear_h3_cache()
|
||||
from meshai import central_normalizer as cn
|
||||
monkeypatch.setattr(cn, "_photon_reverse_places", lambda lat, lon: [])
|
||||
env = {"data": {"adapter": "wzdx", "category": "work_zone.wzdx",
|
||||
"geo": {"centroid": [-116.0, 44.0]},
|
||||
"data": {"road_names": ["SH-1"], "direction": "northbound",
|
||||
"description": "x", "vehicle_impact": "unknown",
|
||||
"end_date": "2026-06-15T18:30:00+00:00",
|
||||
"latitude": 44.0, "longitude": -116.0,
|
||||
"_enriched": {"geocoder": {"city": "Boise"}}}}}
|
||||
n = normalize(env)
|
||||
assert isinstance(n["ends_at"], datetime)
|
||||
assert n["ends_at"].month == 6 and n["ends_at"].day == 15
|
||||
assert n["ends_at"].hour in (18, 11, 12) # depending on local-tz coercion
|
||||
|
||||
|
||||
# --- (e) MM regex extraction on ID description ----------------------------
|
||||
|
||||
def test_wzdx_mile_post_regex_from_description(monkeypatch):
|
||||
_clear_h3_cache()
|
||||
from meshai import central_normalizer as cn
|
||||
monkeypatch.setattr(cn, "_photon_reverse_places", lambda lat, lon: [])
|
||||
env = {"data": {"adapter": "wzdx", "category": "work_zone.wzdx",
|
||||
"geo": {"centroid": [-116.0, 44.0]},
|
||||
"data": {"road_names": ["I-15"], "direction": "southbound",
|
||||
"description": "Bridge work on I-15 SB from MM (89) to MM (93). 6/1/2026 7:00 AM to 6/3/2026 5:00 PM",
|
||||
"vehicle_impact": "some-lanes-closed",
|
||||
"end_date": "2026-06-03T22:00:00Z",
|
||||
"latitude": 44.0, "longitude": -116.0,
|
||||
"_enriched": {"geocoder": {"city": "Blackfoot"}}}}}
|
||||
n = normalize(env)
|
||||
assert n["mile_start"] == 89
|
||||
assert n["mile_end"] == 93
|
||||
|
||||
|
||||
# --- (f) WA event without MM yields mile_start=None -----------------------
|
||||
|
||||
def test_wzdx_wa_no_mm_in_description(monkeypatch):
|
||||
_clear_h3_cache()
|
||||
from meshai import central_normalizer as cn
|
||||
monkeypatch.setattr(cn, "_photon_reverse_places", lambda lat, lon: [])
|
||||
n = _normalize_wzdx(_WZDX_WA)
|
||||
assert n["mile_start"] is None
|
||||
assert n["mile_end"] is None
|
||||
|
||||
|
||||
# --- (g) town fallback chain ----------------------------------------------
|
||||
|
||||
def test_wzdx_town_uses_geocoder_city_when_present(monkeypatch):
|
||||
_clear_h3_cache()
|
||||
from meshai import central_normalizer as cn
|
||||
calls = []
|
||||
monkeypatch.setattr(cn, "_photon_reverse_places",
|
||||
lambda lat, lon: calls.append("called") or [])
|
||||
n = _normalize_wzdx(_WZDX_MCCALL)
|
||||
assert n["town"] == "McCall"
|
||||
assert calls == [] # city present → Photon NOT called
|
||||
|
||||
|
||||
def test_wzdx_town_falls_back_to_nearest_town_when_city_null(monkeypatch):
|
||||
_clear_h3_cache()
|
||||
from meshai import central_normalizer as cn
|
||||
monkeypatch.setattr(cn, "_photon_reverse_places",
|
||||
lambda lat, lon: [
|
||||
{"geometry": {"coordinates": [-117.293, 46.475]},
|
||||
"properties": {"name": "Pomeroy",
|
||||
"osm_key": "place", "osm_value": "city"}},
|
||||
])
|
||||
n = _normalize_wzdx(_WZDX_WA)
|
||||
assert n["town"] == "Pomeroy"
|
||||
|
||||
|
||||
# --- adapter dispatch routes wzdx → _parse_wzdx_federal -------------------
|
||||
|
||||
def test_wzdx_adapter_routes_to_wzdx_parser(monkeypatch):
|
||||
_clear_h3_cache()
|
||||
from meshai import central_normalizer as cn
|
||||
monkeypatch.setattr(cn, "_photon_reverse_places", lambda lat, lon: [])
|
||||
n = normalize(_WZDX_WA)
|
||||
assert n is not None
|
||||
assert n["source"] == "wzdx"
|
||||
|
||||
|
||||
# --- work_type from types_of_work or event_type ---------------------------
|
||||
|
||||
def test_wzdx_sub_type_from_types_of_work(monkeypatch):
|
||||
_clear_h3_cache()
|
||||
from meshai import central_normalizer as cn
|
||||
monkeypatch.setattr(cn, "_photon_reverse_places", lambda lat, lon: [])
|
||||
env = {"data": {"adapter": "wzdx", "category": "work_zone.wzdx",
|
||||
"geo": {"centroid": [-116.0, 44.0]},
|
||||
"data": {"road_names": ["SH-1"], "direction": "both",
|
||||
"description": "x",
|
||||
"types_of_work": [{"type_name": "paving"}],
|
||||
"vehicle_impact": "some-lanes-closed",
|
||||
"end_date": "2026-06-05T17:00:00Z",
|
||||
"latitude": 44.0, "longitude": -116.0,
|
||||
"_enriched": {"geocoder": {"city": "Boise"}}}}}
|
||||
n = normalize(env)
|
||||
# Folded form: impact_phrase + work_type (paving)
|
||||
assert n["sub_type"] == "lanes reduced, paving"
|
||||
|
||||
|
||||
def test_wzdx_sub_type_unknown_vocab_is_lowercased_with_spaces(monkeypatch):
|
||||
_clear_h3_cache()
|
||||
from meshai import central_normalizer as cn
|
||||
monkeypatch.setattr(cn, "_photon_reverse_places", lambda lat, lon: [])
|
||||
env = {"data": {"adapter": "wzdx", "category": "work_zone.wzdx",
|
||||
"geo": {"centroid": [-116.0, 44.0]},
|
||||
"data": {"road_names": ["SH-1"], "direction": "northbound",
|
||||
"description": "x",
|
||||
"types_of_work": [{"type_name": "Some-Custom-Work"}],
|
||||
"vehicle_impact": "all-lanes-open",
|
||||
"end_date": "2026-06-05T17:00:00Z",
|
||||
"latitude": 44.0, "longitude": -116.0,
|
||||
"_enriched": {"geocoder": {"city": "Boise"}}}}}
|
||||
n = normalize(env)
|
||||
assert n["sub_type"] == "some custom work" # lowercased + hyphens→spaces
|
||||
|
|
|
|||
136
tests/test_cold_start_grace.py
Normal file
136
tests/test_cold_start_grace.py
Normal file
|
|
@ -0,0 +1,136 @@
|
|||
"""v0.5.8b — cold-start grace + post-broadcast commit hook in Dispatcher.
|
||||
|
||||
The grace window suppresses mesh broadcasts for N seconds after the FIRST
|
||||
event the dispatcher sees through an enabled toggle. The persistence layer
|
||||
(handler-side) has already run by then, so fires/event_log rows exist;
|
||||
only the broadcast (and the mesh_broadcasts_out audit + last_broadcast_*
|
||||
callback) is gated.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import time
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from meshai.config import Config
|
||||
from meshai.notifications.events import make_event
|
||||
from meshai.notifications.pipeline.dispatcher import Dispatcher
|
||||
|
||||
|
||||
# ---------- helpers -------------------------------------------------------
|
||||
|
||||
|
||||
class RecChannel:
|
||||
def __init__(self, rec):
|
||||
self.rec = rec
|
||||
|
||||
async def deliver(self, payload, rule):
|
||||
self.rec.append({
|
||||
"name": rule.name,
|
||||
"message": payload.message,
|
||||
"delivery_type": rule.delivery_type,
|
||||
})
|
||||
return True
|
||||
|
||||
|
||||
def _cfg(*, cold_start_grace_seconds=60, toggle_name="weather"):
|
||||
cfg = Config()
|
||||
cfg.notifications.rules = []
|
||||
cfg.notifications.cold_start_grace_seconds = cold_start_grace_seconds
|
||||
t = cfg.notifications.toggles[toggle_name]
|
||||
t.enabled = True
|
||||
t.min_severity = "routine"
|
||||
t.severity_channels = {"routine": ["mesh_broadcast"]}
|
||||
# Wide-open v0.5.2 gates so the cold-start gate is the only thing
|
||||
# that can drop these events.
|
||||
t.freshness_seconds = 0
|
||||
t.cooldown_seconds = 0
|
||||
return cfg
|
||||
|
||||
|
||||
def _ev(*, source="nws", category="weather_warning",
|
||||
severity="routine", title="t", **kw):
|
||||
return make_event(
|
||||
source=source, category=category, severity=severity,
|
||||
title=title, timestamp=time.time(), **kw,
|
||||
)
|
||||
|
||||
|
||||
def _make(cfg):
|
||||
rec: list = []
|
||||
d = Dispatcher(cfg, lambda r, c: RecChannel(rec), connector=None)
|
||||
return d, rec
|
||||
|
||||
|
||||
# ---------- (a) first event during grace ----------------------------------
|
||||
|
||||
|
||||
def test_cold_start_grace_drops_first_event_inside_window():
|
||||
cfg = _cfg(cold_start_grace_seconds=60)
|
||||
d, rec = _make(cfg)
|
||||
|
||||
fake_now = 1_000_000.0
|
||||
with patch("meshai.notifications.pipeline.dispatcher.time.time",
|
||||
return_value=fake_now):
|
||||
asyncio.run(d.dispatch(_ev()))
|
||||
|
||||
assert rec == [], "broadcast must be suppressed inside grace window"
|
||||
stats = d.dispatch_stats()
|
||||
assert stats["cold_start_dropped"] == 1
|
||||
assert stats["cold_start_anchor_at"] == fake_now
|
||||
|
||||
|
||||
# ---------- (b) event arriving 30s into grace -- still dropped -----------
|
||||
|
||||
|
||||
def test_cold_start_grace_drops_event_partway_through_window():
|
||||
cfg = _cfg(cold_start_grace_seconds=60)
|
||||
d, rec = _make(cfg)
|
||||
|
||||
t0 = 2_000_000.0
|
||||
# First event anchors the window.
|
||||
with patch("meshai.notifications.pipeline.dispatcher.time.time",
|
||||
return_value=t0):
|
||||
asyncio.run(d.dispatch(_ev()))
|
||||
# 30s in, still inside the 60s window.
|
||||
with patch("meshai.notifications.pipeline.dispatcher.time.time",
|
||||
return_value=t0 + 30):
|
||||
asyncio.run(d.dispatch(_ev()))
|
||||
|
||||
assert rec == []
|
||||
assert d.dispatch_stats()["cold_start_dropped"] == 2
|
||||
|
||||
|
||||
# ---------- (c) event 61s after first -- broadcasts ----------------------
|
||||
|
||||
|
||||
def test_cold_start_grace_passes_event_after_window():
|
||||
cfg = _cfg(cold_start_grace_seconds=60)
|
||||
d, rec = _make(cfg)
|
||||
|
||||
t0 = 3_000_000.0
|
||||
with patch("meshai.notifications.pipeline.dispatcher.time.time",
|
||||
return_value=t0):
|
||||
asyncio.run(d.dispatch(_ev())) # dropped
|
||||
with patch("meshai.notifications.pipeline.dispatcher.time.time",
|
||||
return_value=t0 + 61):
|
||||
asyncio.run(d.dispatch(_ev())) # broadcasts
|
||||
|
||||
assert len(rec) == 1
|
||||
stats = d.dispatch_stats()
|
||||
assert stats["cold_start_dropped"] == 1
|
||||
|
||||
|
||||
# ---------- (d) grace = 0 disables the feature ----------------------------
|
||||
|
||||
|
||||
def test_cold_start_grace_zero_disables_feature():
|
||||
cfg = _cfg(cold_start_grace_seconds=0)
|
||||
d, rec = _make(cfg)
|
||||
asyncio.run(d.dispatch(_ev()))
|
||||
assert len(rec) == 1
|
||||
stats = d.dispatch_stats()
|
||||
assert stats["cold_start_dropped"] == 0
|
||||
# Anchor not set when grace disabled (no gate ran).
|
||||
assert stats["cold_start_anchor_at"] is None
|
||||
|
|
@ -32,6 +32,7 @@ def _dispatch(cfg, event):
|
|||
def _cfg(enable="weather", **kw):
|
||||
cfg = Config()
|
||||
cfg.notifications.rules = []
|
||||
cfg.notifications.cold_start_grace_seconds = 0 # v0.5.8b: legacy tests
|
||||
t = cfg.notifications.toggles[enable]
|
||||
t.enabled = True
|
||||
t.min_severity = kw.get("min_severity", "priority")
|
||||
|
|
@ -47,6 +48,7 @@ def _ev(severity="priority", category="weather_warning", region=None, regions=No
|
|||
|
||||
def test_disabled_toggle_no_dispatch():
|
||||
cfg = Config(); cfg.notifications.rules = [] # weather disabled by default
|
||||
cfg.notifications.cold_start_grace_seconds = 0
|
||||
assert _dispatch(cfg, _ev()) == []
|
||||
|
||||
|
||||
|
|
@ -107,6 +109,7 @@ def test_quiet_hours_override_immediate_only():
|
|||
def test_category_maps_to_correct_family():
|
||||
# seismic family toggle handles earthquake_event via get_toggle fallback
|
||||
cfg = Config(); cfg.notifications.rules = []
|
||||
cfg.notifications.cold_start_grace_seconds = 0 # v0.5.8b: legacy test
|
||||
cfg.notifications.toggles["seismic"].enabled = True
|
||||
cfg.notifications.toggles["seismic"].severity_channels = {"priority": ["mesh_broadcast"]}
|
||||
rec = _dispatch(cfg, _ev(severity="priority", category="earthquake_event"))
|
||||
|
|
|
|||
339
tests/test_persistence.py
Normal file
339
tests/test_persistence.py
Normal file
|
|
@ -0,0 +1,339 @@
|
|||
"""Tests for the meshai/persistence foundation (v1 schema).
|
||||
|
||||
Foundation-only: no adapter handlers wired yet. Tests cover migration
|
||||
runner, schema shape, idempotency, WAL mode, env-var path resolution,
|
||||
in-memory mode, and basic insert/query against representative tables.
|
||||
"""
|
||||
|
||||
import os
|
||||
import sqlite3
|
||||
import tempfile
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from meshai.persistence import (
|
||||
MESHAI_DB_PATH,
|
||||
SCHEMA_VERSION,
|
||||
close_thread_connection,
|
||||
get_db,
|
||||
init_db,
|
||||
)
|
||||
from meshai.persistence import db as persistence_db
|
||||
|
||||
|
||||
# ---------- helpers --------------------------------------------------------
|
||||
|
||||
|
||||
def _force_reinit(monkeypatch, db_path):
|
||||
"""Point every get_db() at a fresh DB and reset the per-process init
|
||||
cache + thread-local connection so tests don't bleed into one another."""
|
||||
monkeypatch.setenv("MESHAI_DB_PATH", db_path)
|
||||
# Wipe the initialised-set so migrations re-run against the new file.
|
||||
persistence_db._initialised.clear()
|
||||
close_thread_connection()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def tmp_db(monkeypatch, tmp_path):
|
||||
db = str(tmp_path / "meshai-test.sqlite")
|
||||
_force_reinit(monkeypatch, db)
|
||||
yield db
|
||||
close_thread_connection()
|
||||
persistence_db._initialised.discard(db)
|
||||
|
||||
|
||||
# ---------- migration runner ----------------------------------------------
|
||||
|
||||
|
||||
# Every table the v1 schema is contracted to produce. The migration runner
|
||||
# is tested by checking that ALL of these exist after init.
|
||||
_V1_TABLES = {
|
||||
"schema_meta",
|
||||
"event_log",
|
||||
"traffic_events",
|
||||
"fires",
|
||||
"firms_pixels",
|
||||
"quake_events",
|
||||
"nws_alerts",
|
||||
"gauge_readings",
|
||||
"swpc_events",
|
||||
"mesh_nodes",
|
||||
"mesh_telemetry",
|
||||
"mesh_positions",
|
||||
"mesh_messages_in",
|
||||
"mesh_broadcasts_out",
|
||||
"mesh_health_events",
|
||||
}
|
||||
|
||||
|
||||
def test_migration_v1_creates_all_tables(tmp_db):
|
||||
conn = init_db()
|
||||
rows = conn.execute(
|
||||
"SELECT name FROM sqlite_master WHERE type='table' "
|
||||
"AND name NOT LIKE 'sqlite_%' AND name NOT LIKE 'idx_%'"
|
||||
).fetchall()
|
||||
names = {r["name"] for r in rows}
|
||||
missing = _V1_TABLES - names
|
||||
extra = names - _V1_TABLES
|
||||
assert not missing, f"missing tables after v1: {sorted(missing)}"
|
||||
# extras are fine (autoindex tables etc.); just make sure we didn't
|
||||
# leave the contract unfulfilled.
|
||||
assert len(names) >= len(_V1_TABLES)
|
||||
|
||||
|
||||
def test_schema_version_recorded(tmp_db):
|
||||
conn = init_db()
|
||||
row = conn.execute("SELECT value FROM schema_meta WHERE key='version'").fetchone()
|
||||
assert row is not None
|
||||
assert int(row["value"]) == SCHEMA_VERSION
|
||||
|
||||
|
||||
def test_migration_idempotent_rerun(tmp_db):
|
||||
init_db()
|
||||
# Force a "second startup" by closing the connection and clearing the
|
||||
# per-process initialised cache so the migration runner re-evaluates.
|
||||
close_thread_connection()
|
||||
persistence_db._initialised.discard(tmp_db)
|
||||
conn = init_db()
|
||||
# No exception means we didn't try to CREATE TABLE twice; verify the
|
||||
# row count of schema_meta hasn't drifted.
|
||||
rows = conn.execute("SELECT COUNT(*) AS n FROM schema_meta").fetchone()
|
||||
assert rows["n"] >= 1
|
||||
|
||||
|
||||
# ---------- WAL mode ------------------------------------------------------
|
||||
|
||||
|
||||
def test_wal_mode_enabled_on_file_db(tmp_db):
|
||||
conn = init_db()
|
||||
mode = conn.execute("PRAGMA journal_mode").fetchone()[0]
|
||||
assert mode.lower() == "wal", f"expected WAL, got {mode!r}"
|
||||
|
||||
|
||||
# ---------- MESHAI_DB_PATH resolution -------------------------------------
|
||||
|
||||
|
||||
def test_meshai_db_path_env_var_respected(monkeypatch, tmp_path):
|
||||
custom = str(tmp_path / "custom-path.sqlite")
|
||||
monkeypatch.setenv("MESHAI_DB_PATH", custom)
|
||||
assert MESHAI_DB_PATH() == custom
|
||||
|
||||
|
||||
def test_meshai_db_path_falls_back_to_default(monkeypatch):
|
||||
monkeypatch.delenv("MESHAI_DB_PATH", raising=False)
|
||||
assert MESHAI_DB_PATH() == persistence_db.DEFAULT_DB_PATH
|
||||
|
||||
|
||||
# ---------- in-memory mode for tests --------------------------------------
|
||||
|
||||
|
||||
def test_in_memory_db_works(monkeypatch):
|
||||
# Special-cased :memory: routes via uri=shared-cache so init runs cleanly.
|
||||
monkeypatch.setenv("MESHAI_DB_PATH", ":memory:")
|
||||
persistence_db._initialised.clear()
|
||||
close_thread_connection()
|
||||
conn = init_db()
|
||||
# All v1 tables present even in memory.
|
||||
rows = conn.execute(
|
||||
"SELECT name FROM sqlite_master WHERE type='table' "
|
||||
"AND name NOT LIKE 'sqlite_%'"
|
||||
).fetchall()
|
||||
names = {r["name"] for r in rows}
|
||||
assert _V1_TABLES.issubset(names)
|
||||
close_thread_connection()
|
||||
|
||||
|
||||
# ---------- basic insert/query: fires -------------------------------------
|
||||
|
||||
|
||||
def test_fires_insert_and_query(tmp_db):
|
||||
conn = init_db()
|
||||
now = int(time.time())
|
||||
irwin = "{E7FCBC00-2D0A-49D6-889F-550D4EDCBFD6}"
|
||||
conn.execute(
|
||||
"INSERT INTO fires(irwin_id, incident_name, incident_type, "
|
||||
"current_acres, current_contained_pct, status, lat, lon, "
|
||||
"county, state, landclass, declared_at, last_event_at) "
|
||||
"VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)",
|
||||
(irwin, "IA 1", "wildfire", None, None, "active",
|
||||
43.5213, -115.1665, "Elmore", "ID", "Sawtooth National Forest",
|
||||
now - 3600, now),
|
||||
)
|
||||
row = conn.execute(
|
||||
"SELECT incident_name, county, state, landclass, last_event_at "
|
||||
"FROM fires WHERE irwin_id = ?", (irwin,)).fetchone()
|
||||
assert row is not None
|
||||
assert row["incident_name"] == "IA 1"
|
||||
assert row["county"] == "Elmore"
|
||||
assert row["state"] == "ID"
|
||||
assert row["landclass"] == "Sawtooth National Forest"
|
||||
assert row["last_event_at"] == now
|
||||
|
||||
|
||||
def test_fires_last_broadcast_change_detection_columns(tmp_db):
|
||||
"""The 8h-rate-limit change-detection logic uses last_broadcast_acres
|
||||
and last_broadcast_contained -- verify they accept NULL and updates."""
|
||||
conn = init_db()
|
||||
now = int(time.time())
|
||||
irwin = "{ABC}"
|
||||
conn.execute(
|
||||
"INSERT INTO fires(irwin_id, incident_name, last_event_at, "
|
||||
"last_broadcast_at, last_broadcast_acres, last_broadcast_contained) "
|
||||
"VALUES (?,?,?,?,?,?)",
|
||||
(irwin, "X", now, now, 1847.0, 23),
|
||||
)
|
||||
row = conn.execute(
|
||||
"SELECT last_broadcast_acres, last_broadcast_contained "
|
||||
"FROM fires WHERE irwin_id = ?", (irwin,)).fetchone()
|
||||
assert row["last_broadcast_acres"] == 1847.0
|
||||
assert row["last_broadcast_contained"] == 23
|
||||
|
||||
|
||||
# ---------- basic insert/query: mesh_nodes -------------------------------
|
||||
|
||||
|
||||
def test_mesh_nodes_insert_and_query(tmp_db):
|
||||
conn = init_db()
|
||||
now = int(time.time())
|
||||
conn.execute(
|
||||
"INSERT INTO mesh_nodes(node_id, long_name, short_name, hw_model, "
|
||||
"last_lat, last_lon, last_battery_pct, first_seen_at, last_seen_at) "
|
||||
"VALUES (?,?,?,?,?,?,?,?,?)",
|
||||
("!85098cea", "AIDA Northgate", "AIDA", "TBEAM",
|
||||
43.6535, -116.2674, 88, now - 86400, now),
|
||||
)
|
||||
row = conn.execute(
|
||||
"SELECT long_name, short_name, last_battery_pct, is_stale "
|
||||
"FROM mesh_nodes WHERE node_id = ?", ("!85098cea",)).fetchone()
|
||||
assert row is not None
|
||||
assert row["long_name"] == "AIDA Northgate"
|
||||
assert row["last_battery_pct"] == 88
|
||||
assert row["is_stale"] == 0 # default
|
||||
|
||||
|
||||
# ---------- traffic_events composite PK ----------------------------------
|
||||
|
||||
|
||||
def test_traffic_events_composite_pk_uniqueness(tmp_db):
|
||||
conn = init_db()
|
||||
now = int(time.time())
|
||||
conn.execute(
|
||||
"INSERT INTO traffic_events(source, external_id, road, county, state, "
|
||||
"first_seen_at, last_seen_at) VALUES (?,?,?,?,?,?,?)",
|
||||
("state_511_atis", "ID:Construction:33868", "I-86", "Bannock", "ID", now, now),
|
||||
)
|
||||
# Same source + external_id should violate the PK constraint.
|
||||
with pytest.raises(sqlite3.IntegrityError):
|
||||
conn.execute(
|
||||
"INSERT INTO traffic_events(source, external_id, road, county, state, "
|
||||
"first_seen_at, last_seen_at) VALUES (?,?,?,?,?,?,?)",
|
||||
("state_511_atis", "ID:Construction:33868", "I-86", "Bannock", "ID", now, now),
|
||||
)
|
||||
# Different source with same external_id is fine.
|
||||
conn.execute(
|
||||
"INSERT INTO traffic_events(source, external_id, road, county, state, "
|
||||
"first_seen_at, last_seen_at) VALUES (?,?,?,?,?,?,?)",
|
||||
("wzdx", "ID:Construction:33868", "I-86", "Bannock", "ID", now, now),
|
||||
)
|
||||
rows = conn.execute("SELECT COUNT(*) AS n FROM traffic_events").fetchone()
|
||||
assert rows["n"] == 2
|
||||
|
||||
|
||||
# ---------- event_log basic ------------------------------------------------
|
||||
|
||||
|
||||
def test_event_log_insert_and_query(tmp_db):
|
||||
conn = init_db()
|
||||
now = int(time.time())
|
||||
conn.execute(
|
||||
"INSERT INTO event_log(received_at, source, category, severity_word, "
|
||||
"event_id_external, nats_subject, handled, table_name, table_pk) "
|
||||
"VALUES (?,?,?,?,?,?,?,?,?)",
|
||||
(now, "wfigs_incidents", "fire.incident.wildfire", "routine",
|
||||
"{E7FCBC00}", "central.fire.incident.id.elmore", 1, "fires", "{E7FCBC00}"),
|
||||
)
|
||||
row = conn.execute(
|
||||
"SELECT source, table_name, handled FROM event_log "
|
||||
"WHERE event_id_external = ?", ("{E7FCBC00}",)).fetchone()
|
||||
assert row["source"] == "wfigs_incidents"
|
||||
assert row["table_name"] == "fires"
|
||||
assert row["handled"] == 1
|
||||
|
||||
|
||||
# ---------- firms_pixels nullable FK to fires ----------------------------
|
||||
|
||||
|
||||
def test_firms_pixels_fk_to_fires_nullable(tmp_db):
|
||||
conn = init_db()
|
||||
now = int(time.time())
|
||||
# Unattached pixel (no irwin_id yet -- v0.6 fire-tracker will attach later).
|
||||
conn.execute(
|
||||
"INSERT INTO firms_pixels(lat, lon, acq_time, frp, confidence, satellite) "
|
||||
"VALUES (?,?,?,?,?,?)",
|
||||
(42.197, -113.710, now, 135.93, "high", "N"),
|
||||
)
|
||||
row = conn.execute(
|
||||
"SELECT irwin_id, frp, confidence FROM firms_pixels "
|
||||
"WHERE acq_time = ?", (now,)).fetchone()
|
||||
assert row is not None
|
||||
assert row["irwin_id"] is None # nullable
|
||||
assert row["frp"] == 135.93
|
||||
|
||||
# Insert a fire, then attach a new pixel via the FK.
|
||||
irwin = "{LINK-IRWIN}"
|
||||
conn.execute(
|
||||
"INSERT INTO fires(irwin_id, incident_name, last_event_at) "
|
||||
"VALUES (?,?,?)", (irwin, "Linked Fire", now),
|
||||
)
|
||||
conn.execute(
|
||||
"INSERT INTO firms_pixels(irwin_id, lat, lon, acq_time, frp) "
|
||||
"VALUES (?,?,?,?,?)", (irwin, 42.197, -113.710, now + 1, 50.0),
|
||||
)
|
||||
rows = conn.execute(
|
||||
"SELECT COUNT(*) AS n FROM firms_pixels WHERE irwin_id = ?",
|
||||
(irwin,)).fetchone()
|
||||
assert rows["n"] == 1
|
||||
|
||||
|
||||
# ---------- indexes present (spot-check) ---------------------------------
|
||||
|
||||
|
||||
def test_v1_indexes_created(tmp_db):
|
||||
conn = init_db()
|
||||
rows = conn.execute(
|
||||
"SELECT name FROM sqlite_master WHERE type='index' "
|
||||
"AND name LIKE 'idx_%'"
|
||||
).fetchall()
|
||||
names = {r["name"] for r in rows}
|
||||
must_have = {
|
||||
"idx_event_log_received",
|
||||
"idx_traffic_last_seen",
|
||||
"idx_fires_last_event",
|
||||
"idx_firms_pixels_acq_time",
|
||||
"idx_mesh_tel_node_time",
|
||||
"idx_mesh_pos_node_time",
|
||||
"idx_mesh_nodes_last_seen",
|
||||
"idx_gauge_site_time",
|
||||
}
|
||||
missing = must_have - names
|
||||
assert not missing, f"missing indexes: {sorted(missing)}"
|
||||
|
||||
|
||||
# ---------- gauge_readings autoincrement -------------------------------
|
||||
|
||||
|
||||
def test_gauge_readings_autoincrement_pk(tmp_db):
|
||||
conn = init_db()
|
||||
now = int(time.time())
|
||||
for value in (3490.0, 3520.0, 3505.0):
|
||||
conn.execute(
|
||||
"INSERT INTO gauge_readings(site_id, reading_value, reading_unit, "
|
||||
"reading_time) VALUES (?,?,?,?)",
|
||||
("USGS-13038000", value, "ft^3/s", now),
|
||||
)
|
||||
rows = conn.execute(
|
||||
"SELECT COUNT(*) AS n FROM gauge_readings WHERE site_id = ?",
|
||||
("USGS-13038000",)).fetchone()
|
||||
assert rows["n"] == 3
|
||||
|
|
@ -58,6 +58,10 @@ def _cfg(toggle_name="weather", **kw):
|
|||
"""Default config: one toggle enabled with mesh_broadcast on priority."""
|
||||
cfg = Config()
|
||||
cfg.notifications.rules = []
|
||||
# v0.5.8b: disable the cold-start grace for these tests -- they
|
||||
# exercise the v0.5.2 guards in isolation and expect the first
|
||||
# event to broadcast.
|
||||
cfg.notifications.cold_start_grace_seconds = 0
|
||||
t = cfg.notifications.toggles[toggle_name]
|
||||
t.enabled = True
|
||||
t.min_severity = kw.get("min_severity", "routine")
|
||||
|
|
@ -280,6 +284,7 @@ def test_hydro_event_maps_to_geohazards_toggle():
|
|||
toggle alone must NOT fire on them anymore."""
|
||||
cfg = Config()
|
||||
cfg.notifications.rules = []
|
||||
cfg.notifications.cold_start_grace_seconds = 0
|
||||
# Enable BOTH weather and seismic toggles so we can prove routing.
|
||||
cfg.notifications.toggles["weather"].enabled = True
|
||||
cfg.notifications.toggles["weather"].min_severity = "routine"
|
||||
|
|
@ -308,6 +313,7 @@ def test_hydro_high_water_also_seismic():
|
|||
"""Same as above for stream_high_water (the lower-severity sibling)."""
|
||||
cfg = Config()
|
||||
cfg.notifications.rules = []
|
||||
cfg.notifications.cold_start_grace_seconds = 0
|
||||
cfg.notifications.toggles["seismic"].enabled = True
|
||||
cfg.notifications.toggles["seismic"].min_severity = "routine"
|
||||
cfg.notifications.toggles["seismic"].severity_channels = {
|
||||
|
|
@ -331,5 +337,6 @@ def test_dispatch_stats_exposes_all_counters():
|
|||
stats = d.dispatch_stats()
|
||||
assert set(stats.keys()) == {
|
||||
"stale_dropped", "cooldown_dropped", "dedup_dropped",
|
||||
"cold_start_dropped", "cold_start_anchor_at",
|
||||
"cooldown_keys", "dedup_lru_size",
|
||||
}
|
||||
|
|
|
|||
589
tests/test_wfigs_handler.py
Normal file
589
tests/test_wfigs_handler.py
Normal file
|
|
@ -0,0 +1,589 @@
|
|||
"""Tests for meshai/central/wfigs_handler.py -- WFIGS persistence wire-up.
|
||||
|
||||
Covers:
|
||||
(a) parse clean active-incident envelope (all fields populated)
|
||||
(b) acres fallback chain: top-null -> raw.DiscoveryAcres used
|
||||
(c) acres absent at every level -> renders "N/A"
|
||||
(d) IncidentName="IA 1" placeholder passes through verbatim
|
||||
(e) tombstone subject -> handler returns None + event_log row handled=0
|
||||
(f) perimeter subject -> handler returns None + event_log row handled=0
|
||||
(g) NEW IRWIN -> "New:" prefix + fires INSERT + mesh_broadcasts_out audit row
|
||||
(h) known IRWIN no change -> drop silently, last_broadcast_* unchanged
|
||||
(i) known IRWIN acres up but <8h elapsed -> drop, last_broadcast_* unchanged
|
||||
(j) known IRWIN acres up + >=8h elapsed -> "Update:" prefix + audit row
|
||||
(k) location anchor priority: geocoder.city > nearest_town > landclass > county
|
||||
"""
|
||||
|
||||
import os
|
||||
import time
|
||||
|
||||
import pytest
|
||||
|
||||
from meshai import central_normalizer as cn
|
||||
from meshai.central.wfigs_handler import (
|
||||
WFIGS_BROADCAST_COOLDOWN_S,
|
||||
handle_wfigs,
|
||||
)
|
||||
from meshai.persistence import close_thread_connection, init_db
|
||||
from meshai.persistence import db as persistence_db
|
||||
|
||||
|
||||
# ---------- fixtures ------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mem_db(monkeypatch, tmp_path):
|
||||
"""Fresh on-disk SQLite per test (avoids in-memory shared-cache bleed)."""
|
||||
db_path = str(tmp_path / "wfigs-test.sqlite")
|
||||
monkeypatch.setenv("MESHAI_DB_PATH", db_path)
|
||||
persistence_db._initialised.clear()
|
||||
close_thread_connection()
|
||||
conn = init_db()
|
||||
yield conn
|
||||
close_thread_connection()
|
||||
persistence_db._initialised.discard(db_path)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def no_photon(monkeypatch):
|
||||
"""Force nearest_town to return None so anchor falls through deterministically.
|
||||
Tests that exercise nearest_town wire it in directly."""
|
||||
monkeypatch.setattr(cn, "_photon_reverse_places", lambda lat, lon: [])
|
||||
# Also reset the H3 LRU so cache state doesn't leak across tests.
|
||||
if hasattr(cn, "_H3_NEAREST_CACHE"):
|
||||
cn._H3_NEAREST_CACHE.clear()
|
||||
|
||||
|
||||
# ---------- envelope builders --------------------------------------------
|
||||
|
||||
|
||||
_IRWIN_A = "{E7FCBC00-2D0A-49D6-889F-550D4EDCBFD6}"
|
||||
_IRWIN_B = "{ABCDEF01-2345-6789-ABCD-EF0123456789}"
|
||||
_IRWIN_C = "{11111111-2222-3333-4444-555555555555}"
|
||||
|
||||
|
||||
def _make_active_envelope(*, irwin_id=_IRWIN_A,
|
||||
name="Cache Peak Fire",
|
||||
incident_type="wildfire",
|
||||
lat=42.197, lon=-113.710,
|
||||
county="Cassia", state="ID",
|
||||
landclass=None,
|
||||
geocoder_city=None,
|
||||
daily_acres=1847.0,
|
||||
pct_contained=23,
|
||||
raw_discovery_acres=None,
|
||||
raw_pct_contained=None,
|
||||
fire_discovery_dt_ms=1780529163000,
|
||||
subject="central.fire.incident.id.cassia"):
|
||||
"""Build the Central CloudEvents envelope shape we observe in prod."""
|
||||
geocoder = {}
|
||||
if geocoder_city is not None:
|
||||
geocoder["city"] = geocoder_city
|
||||
if landclass is not None:
|
||||
geocoder["landclass"] = landclass
|
||||
raw = {}
|
||||
if raw_discovery_acres is not None:
|
||||
raw["DiscoveryAcres"] = raw_discovery_acres
|
||||
if raw_pct_contained is not None:
|
||||
raw["PercentContained"] = raw_pct_contained
|
||||
return {
|
||||
"subject": subject,
|
||||
"id": f"{irwin_id}:active:{int(time.time())}",
|
||||
"data": {
|
||||
"id": irwin_id,
|
||||
"adapter": "wfigs_incidents",
|
||||
"category": f"fire.incident.{incident_type}",
|
||||
"severity": "routine",
|
||||
"geo": {
|
||||
"primary_region": f"US-{state}",
|
||||
"centroid": [lon, lat],
|
||||
"geocoder": geocoder,
|
||||
},
|
||||
"data": {
|
||||
"IrwinID": irwin_id,
|
||||
"IncidentName": name,
|
||||
"IncidentTypeCategory": incident_type,
|
||||
"latitude": lat,
|
||||
"longitude": lon,
|
||||
"POOCounty": county,
|
||||
"POOState": state,
|
||||
"DailyAcres": daily_acres,
|
||||
"PercentContained": pct_contained,
|
||||
"FireDiscoveryDateTime": fire_discovery_dt_ms,
|
||||
"raw": raw,
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _make_tombstone(irwin_id=_IRWIN_A, state="ID", county="Boise",
|
||||
subject="central.fire.incident.removed.id"):
|
||||
return {
|
||||
"subject": subject,
|
||||
"id": f"{irwin_id}:removed:2026-06-04T02:57:04.684858+00:00",
|
||||
"data": {
|
||||
"id": f"{irwin_id}:removed:2026-06-04T02:57:04.684858+00:00",
|
||||
"adapter": "wfigs_incidents",
|
||||
"category": "fire.incident.removed",
|
||||
"severity": "routine",
|
||||
"geo": {"primary_region": f"US-{state}", "geocoder": {}},
|
||||
"data": {
|
||||
"irwin_id": irwin_id,
|
||||
"last_observed_at": "2026-06-04T02:52:04.209539+00:00",
|
||||
"state": state,
|
||||
"county": county,
|
||||
"reason": "fallen_off_current_service",
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _make_perimeter(irwin_id=_IRWIN_A, state="ID", county="Cassia",
|
||||
subject="central.fire.perimeter.id.cassia"):
|
||||
return {
|
||||
"subject": subject,
|
||||
"id": f"{irwin_id}:perimeter",
|
||||
"data": {
|
||||
"id": f"{irwin_id}:perimeter",
|
||||
"adapter": "wfigs_perimeters",
|
||||
"category": "fire.perimeter.wildfire",
|
||||
"severity": "routine",
|
||||
"geo": {"primary_region": f"US-{state}", "geocoder": {}},
|
||||
"data": {
|
||||
"irwin_id": irwin_id,
|
||||
"state": state,
|
||||
"county": county,
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# (a) parse a clean active-incident envelope with all fields
|
||||
# ============================================================================
|
||||
def test_a_parse_clean_active_envelope(mem_db, no_photon):
|
||||
env = _make_active_envelope()
|
||||
n = cn.normalize(env)
|
||||
assert n is not None
|
||||
assert n["_kind"] == "wfigs_incident"
|
||||
assert n["irwin_id"] == _IRWIN_A
|
||||
assert n["incident_name"] == "Cache Peak Fire"
|
||||
assert n["incident_type"] == "wildfire"
|
||||
assert n["acres"] == 1847.0
|
||||
assert n["contained_pct"] == 23
|
||||
assert n["county"] == "Cassia"
|
||||
assert n["state"] == "ID"
|
||||
# FireDiscoveryDateTime epoch-ms -> epoch-s conversion
|
||||
assert n["declared_at_epoch"] == 1780529163
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# (b) null top-level acres -> raw.DiscoveryAcres fallback used
|
||||
# ============================================================================
|
||||
def test_b_acres_fallback_to_raw_discovery_acres(mem_db, no_photon):
|
||||
env = _make_active_envelope(daily_acres=None, pct_contained=None,
|
||||
raw_discovery_acres=0.1,
|
||||
raw_pct_contained=0)
|
||||
n = cn.normalize(env)
|
||||
assert n["acres"] == 0.1
|
||||
assert n["contained_pct"] == 0
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# (c) no acres anywhere -> renders "N/A"
|
||||
# ============================================================================
|
||||
def test_c_acres_missing_renders_na(mem_db, no_photon):
|
||||
env = _make_active_envelope(name="IA 7", daily_acres=None,
|
||||
pct_contained=None,
|
||||
irwin_id=_IRWIN_C,
|
||||
landclass="Sawtooth National Forest")
|
||||
wire = handle_wfigs(cn.normalize(env), env, env["subject"], now=1_000_000)
|
||||
assert wire is not None
|
||||
assert "N/A" in wire
|
||||
assert "containment unknown" in wire
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# (d) "IA 1" placeholder name passes through verbatim
|
||||
# ============================================================================
|
||||
def test_d_ia_placeholder_passthrough(mem_db, no_photon):
|
||||
env = _make_active_envelope(name="IA 1", county="Elmore",
|
||||
daily_acres=None, pct_contained=None,
|
||||
landclass="Sawtooth National Forest",
|
||||
irwin_id=_IRWIN_B)
|
||||
wire = handle_wfigs(cn.normalize(env), env, env["subject"], now=1_000_000)
|
||||
assert wire is not None
|
||||
assert "IA 1" in wire
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# (e) tombstone subject -> handler returns None + event_log handled=0
|
||||
# ============================================================================
|
||||
def test_e_tombstone_returns_none_and_logs(mem_db, no_photon):
|
||||
env = _make_tombstone()
|
||||
n = cn.normalize(env)
|
||||
assert n["_kind"] == "wfigs_tombstone"
|
||||
out = handle_wfigs(n, env, env["subject"], now=2_000_000)
|
||||
assert out is None
|
||||
row = mem_db.execute(
|
||||
"SELECT source, category, handled, table_name, table_pk, nats_subject "
|
||||
"FROM event_log WHERE event_id_external=?", (_IRWIN_A,)).fetchone()
|
||||
assert row is not None
|
||||
assert row["source"] == "wfigs_incidents"
|
||||
assert row["category"] == "fire.incident.removed"
|
||||
assert row["handled"] == 0
|
||||
assert row["table_name"] is None
|
||||
assert row["table_pk"] == _IRWIN_A
|
||||
assert row["nats_subject"] == "central.fire.incident.removed.id"
|
||||
# No row in fires.
|
||||
n_fires = mem_db.execute("SELECT COUNT(*) AS n FROM fires").fetchone()["n"]
|
||||
assert n_fires == 0
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# (f) perimeter subject -> same as tombstone
|
||||
# ============================================================================
|
||||
def test_f_perimeter_returns_none_and_logs(mem_db, no_photon):
|
||||
env = _make_perimeter()
|
||||
n = cn.normalize(env)
|
||||
assert n["_kind"] == "wfigs_perimeter"
|
||||
out = handle_wfigs(n, env, env["subject"], now=3_000_000)
|
||||
assert out is None
|
||||
row = mem_db.execute(
|
||||
"SELECT source, handled FROM event_log WHERE event_id_external=?",
|
||||
(_IRWIN_A,)).fetchone()
|
||||
assert row is not None
|
||||
assert row["source"] == "wfigs_perimeters"
|
||||
assert row["handled"] == 0
|
||||
n_fires = mem_db.execute("SELECT COUNT(*) AS n FROM fires").fetchone()["n"]
|
||||
assert n_fires == 0
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# (g) NEW IRWIN -> "New:" prefix + fires INSERT + mesh_broadcasts_out audit
|
||||
# ============================================================================
|
||||
def test_g_new_irwin_inserts_and_broadcasts(mem_db, no_photon):
|
||||
env = _make_active_envelope(geocoder_city="Burley") # avoids Photon path
|
||||
now = 5_000_000
|
||||
data = {}
|
||||
wire = handle_wfigs(cn.normalize(env), env, env["subject"],
|
||||
data=data, now=now)
|
||||
assert wire is not None
|
||||
assert wire.startswith("🔥 New: Cache Peak Fire")
|
||||
assert "Burley" in wire
|
||||
assert "1,847 ac" in wire
|
||||
assert "23% contained" in wire
|
||||
assert "@ 42.197,-113.710" in wire
|
||||
|
||||
# v0.5.8b: handler INSERTs the fires row with last_broadcast_*=NULL,
|
||||
# then attaches a commit callback. The dispatcher fires the callback
|
||||
# on successful broadcast; we simulate that here.
|
||||
fr_pre = mem_db.execute(
|
||||
"SELECT last_broadcast_at FROM fires WHERE irwin_id=?",
|
||||
(_IRWIN_A,)).fetchone()
|
||||
assert fr_pre["last_broadcast_at"] is None
|
||||
data["_on_broadcast_committed"](float(now))
|
||||
fr = mem_db.execute(
|
||||
"SELECT current_acres, last_broadcast_at, last_broadcast_acres, "
|
||||
"last_broadcast_contained, last_event_at "
|
||||
"FROM fires WHERE irwin_id=?", (_IRWIN_A,)).fetchone()
|
||||
assert fr is not None
|
||||
assert fr["current_acres"] == 1847.0
|
||||
assert fr["last_broadcast_at"] == now
|
||||
assert fr["last_broadcast_acres"] == 1847.0
|
||||
assert fr["last_broadcast_contained"] == 23
|
||||
assert fr["last_event_at"] == now
|
||||
|
||||
# event_log row logged with handled=1.
|
||||
el = mem_db.execute(
|
||||
"SELECT handled, table_name, table_pk FROM event_log "
|
||||
"WHERE event_id_external=?", (_IRWIN_A,)).fetchone()
|
||||
assert el["handled"] == 1
|
||||
assert el["table_name"] == "fires"
|
||||
assert el["table_pk"] == _IRWIN_A
|
||||
|
||||
# v0.5.8b: mesh_broadcasts_out is inserted by the dispatcher
|
||||
# (test_cold_start_grace covers that path). The handler only signals
|
||||
# via data["_broadcast_audit"] that an audit row is wanted.
|
||||
assert data["_broadcast_audit"] == {"table": "fires", "pk": _IRWIN_A}
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# (h) known IRWIN no-change -> drop silently, last_broadcast_* unchanged
|
||||
# ============================================================================
|
||||
def test_h_known_irwin_no_change_drops(mem_db, no_photon):
|
||||
env = _make_active_envelope(geocoder_city="Burley")
|
||||
first_now = 5_000_000
|
||||
data0 = {}
|
||||
handle_wfigs(cn.normalize(env), env, env["subject"],
|
||||
data=data0, now=first_now)
|
||||
# v0.5.8b: dispatcher commit closes the broadcast.
|
||||
data0["_on_broadcast_committed"](float(first_now))
|
||||
|
||||
# Re-publish the same incident exactly 30 min later: same acres + contained.
|
||||
later = first_now + 1800
|
||||
out = handle_wfigs(cn.normalize(env), env, env["subject"], now=later)
|
||||
assert out is None
|
||||
|
||||
fr = mem_db.execute(
|
||||
"SELECT last_broadcast_at, last_broadcast_acres, last_broadcast_contained, "
|
||||
"last_event_at FROM fires WHERE irwin_id=?", (_IRWIN_A,)).fetchone()
|
||||
# last_broadcast_* unchanged from the original.
|
||||
assert fr["last_broadcast_at"] == first_now
|
||||
assert fr["last_broadcast_acres"] == 1847.0
|
||||
assert fr["last_broadcast_contained"] == 23
|
||||
# last_event_at was refreshed.
|
||||
assert fr["last_event_at"] == later
|
||||
|
||||
# v0.5.8b: mesh_broadcasts_out is inserted by the dispatcher, not the
|
||||
# handler -- this test never invokes a real dispatcher, so count is 0.
|
||||
cnt = mem_db.execute(
|
||||
"SELECT COUNT(*) AS n FROM mesh_broadcasts_out WHERE source_event_pk=?",
|
||||
(_IRWIN_A,)).fetchone()["n"]
|
||||
assert cnt == 0
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# (i) known IRWIN acres up but <8h elapsed -> drop, last_broadcast_* unchanged
|
||||
# ============================================================================
|
||||
def test_i_known_irwin_change_inside_cooldown_drops(mem_db, no_photon):
|
||||
env_initial = _make_active_envelope(geocoder_city="Burley")
|
||||
data0 = {}
|
||||
handle_wfigs(cn.normalize(env_initial), env_initial,
|
||||
env_initial["subject"], data=data0, now=5_000_000)
|
||||
data0["_on_broadcast_committed"](float(5_000_000))
|
||||
|
||||
# Bigger fire, but only 4h later -- inside cooldown.
|
||||
env_grown = _make_active_envelope(geocoder_city="Burley",
|
||||
daily_acres=3000.0, pct_contained=23)
|
||||
later = 5_000_000 + 4 * 3600
|
||||
out = handle_wfigs(cn.normalize(env_grown), env_grown,
|
||||
env_grown["subject"], now=later)
|
||||
assert out is None
|
||||
|
||||
fr = mem_db.execute(
|
||||
"SELECT last_broadcast_at, last_broadcast_acres, last_broadcast_contained, "
|
||||
"current_acres FROM fires WHERE irwin_id=?", (_IRWIN_A,)).fetchone()
|
||||
assert fr["last_broadcast_at"] == 5_000_000
|
||||
assert fr["last_broadcast_acres"] == 1847.0
|
||||
assert fr["last_broadcast_contained"] == 23
|
||||
# current_acres was refreshed to the new value.
|
||||
assert fr["current_acres"] == 3000.0
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# (j) known IRWIN acres up AND >=8h elapsed -> "Update:" + audit row
|
||||
# ============================================================================
|
||||
def test_j_known_irwin_change_after_cooldown_broadcasts(mem_db, no_photon):
|
||||
env_initial = _make_active_envelope(geocoder_city="Burley")
|
||||
data_j0 = {}
|
||||
handle_wfigs(cn.normalize(env_initial), env_initial,
|
||||
env_initial["subject"], data=data_j0, now=5_000_000)
|
||||
data_j0["_on_broadcast_committed"](float(5_000_000))
|
||||
|
||||
env_grown = _make_active_envelope(geocoder_city="Burley",
|
||||
daily_acres=3000.0, pct_contained=35)
|
||||
later = 5_000_000 + WFIGS_BROADCAST_COOLDOWN_S
|
||||
data2 = {}
|
||||
out = handle_wfigs(cn.normalize(env_grown), env_grown,
|
||||
env_grown["subject"], data=data2, now=later)
|
||||
assert out is not None
|
||||
assert out.startswith("🔥 Update: Cache Peak Fire")
|
||||
assert "3,000 ac" in out
|
||||
assert "35% contained" in out
|
||||
|
||||
# Simulate dispatcher commit.
|
||||
data2["_on_broadcast_committed"](float(later))
|
||||
fr = mem_db.execute(
|
||||
"SELECT last_broadcast_at, last_broadcast_acres, last_broadcast_contained "
|
||||
"FROM fires WHERE irwin_id=?", (_IRWIN_A,)).fetchone()
|
||||
assert fr["last_broadcast_at"] == later
|
||||
assert fr["last_broadcast_acres"] == 3000.0
|
||||
assert fr["last_broadcast_contained"] == 35
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# (k) location anchor priority -- city > nearest_town > landclass > county
|
||||
# ============================================================================
|
||||
def test_k_anchor_geocoder_city_wins(mem_db, no_photon):
|
||||
env = _make_active_envelope(geocoder_city="Twin Falls",
|
||||
landclass="Sawtooth NF",
|
||||
county="Cassia")
|
||||
wire = handle_wfigs(cn.normalize(env), env, env["subject"], now=1)
|
||||
assert "Twin Falls" in wire
|
||||
assert "Sawtooth NF" not in wire
|
||||
assert "Cassia Co" not in wire
|
||||
|
||||
|
||||
def test_k_anchor_falls_to_nearest_town(monkeypatch, mem_db):
|
||||
"""When city missing, nearest_town(distance, bearing) feeds the anchor."""
|
||||
fake = {"name": "Boise", "distance_mi": 47.0, "bearing": "S"}
|
||||
monkeypatch.setattr(
|
||||
"meshai.central_normalizer.nearest_town",
|
||||
lambda lat, lon, max_distance_mi=50.0: fake,
|
||||
)
|
||||
env = _make_active_envelope(geocoder_city=None,
|
||||
landclass="Sawtooth NF",
|
||||
county="Cassia")
|
||||
wire = handle_wfigs(cn.normalize(env), env, env["subject"], now=1)
|
||||
assert "47 mi S of Boise" in wire
|
||||
# Lower-priority anchors NOT used when nearest_town hit.
|
||||
assert "Sawtooth NF" not in wire
|
||||
|
||||
|
||||
def test_k_anchor_falls_to_landclass(monkeypatch, mem_db):
|
||||
monkeypatch.setattr(
|
||||
"meshai.central_normalizer.nearest_town",
|
||||
lambda lat, lon, max_distance_mi=50.0: None,
|
||||
)
|
||||
env = _make_active_envelope(geocoder_city=None,
|
||||
landclass="Sawtooth National Forest",
|
||||
county="Cassia")
|
||||
wire = handle_wfigs(cn.normalize(env), env, env["subject"], now=1)
|
||||
assert "Sawtooth National Forest" in wire
|
||||
assert "Cassia Co" not in wire
|
||||
|
||||
|
||||
def test_k_anchor_falls_to_county(monkeypatch, mem_db):
|
||||
monkeypatch.setattr(
|
||||
"meshai.central_normalizer.nearest_town",
|
||||
lambda lat, lon, max_distance_mi=50.0: None,
|
||||
)
|
||||
env = _make_active_envelope(geocoder_city=None, landclass=None,
|
||||
county="Cassia", state="ID")
|
||||
wire = handle_wfigs(cn.normalize(env), env, env["subject"], now=1)
|
||||
assert "Cassia Co ID" in wire
|
||||
|
||||
|
||||
def test_k_anchor_nearest_town_under_one_mile_says_near(monkeypatch, mem_db):
|
||||
fake = {"name": "Burley", "distance_mi": 0.3, "bearing": "N"}
|
||||
monkeypatch.setattr(
|
||||
"meshai.central_normalizer.nearest_town",
|
||||
lambda lat, lon, max_distance_mi=50.0: fake,
|
||||
)
|
||||
env = _make_active_envelope(geocoder_city=None)
|
||||
wire = handle_wfigs(cn.normalize(env), env, env["subject"], now=1)
|
||||
assert "near Burley" in wire
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# v0.5.8b refactor -- New:/Update: prefix survives cold-start drops
|
||||
# ============================================================================
|
||||
|
||||
|
||||
def _run_handler_only(env, data=None, now=None):
|
||||
"""Run normalize + handler WITHOUT invoking any commit callback.
|
||||
Simulates the dispatcher dropping the broadcast (grace/cooldown/stale)
|
||||
after the handler has already written persistence rows."""
|
||||
n = cn.normalize(env)
|
||||
if data is None:
|
||||
data = {}
|
||||
wire = handle_wfigs(n, env, env["subject"], data=data, now=now)
|
||||
return wire, data
|
||||
|
||||
|
||||
def _commit(data, committed_at):
|
||||
"""Simulate the dispatcher invoking the handler's post-commit callback."""
|
||||
cb = data.get("_on_broadcast_committed")
|
||||
assert callable(cb), "handler must attach _on_broadcast_committed"
|
||||
cb(committed_at)
|
||||
|
||||
|
||||
def test_e_cold_start_then_resume_still_new(mem_db, no_photon):
|
||||
"""Cold-start drop scenario: first pass writes fires + event_log but
|
||||
dispatcher drops the broadcast (we skip the callback). Second pass on
|
||||
the SAME IRWIN must still produce "New:" because last_broadcast_at is
|
||||
still NULL -- it really is the first delivery for that fire.
|
||||
"""
|
||||
env = _make_active_envelope(geocoder_city="Burley")
|
||||
|
||||
# Pass 1: handler runs, but the dispatcher drops the broadcast (we
|
||||
# mimic that by not calling the commit callback).
|
||||
wire1, data1 = _run_handler_only(env, now=10_000)
|
||||
assert wire1.startswith("🔥 New: ")
|
||||
fr = mem_db.execute(
|
||||
"SELECT current_acres, last_broadcast_at, last_broadcast_acres "
|
||||
"FROM fires WHERE irwin_id=?", (_IRWIN_A,)).fetchone()
|
||||
assert fr is not None
|
||||
assert fr["current_acres"] == 1847.0
|
||||
assert fr["last_broadcast_at"] is None
|
||||
assert fr["last_broadcast_acres"] is None
|
||||
|
||||
# Pass 2: same envelope 5 minutes later (still pre-broadcast).
|
||||
wire2, data2 = _run_handler_only(env, now=10_300)
|
||||
assert wire2.startswith("🔥 New: "), "must still be 'New:' until last_broadcast_at gets set"
|
||||
|
||||
fr2 = mem_db.execute(
|
||||
"SELECT current_acres, last_broadcast_at, last_event_at "
|
||||
"FROM fires WHERE irwin_id=?", (_IRWIN_A,)).fetchone()
|
||||
# last_event_at advanced; last_broadcast_at still NULL.
|
||||
assert fr2["last_event_at"] == 10_300
|
||||
assert fr2["last_broadcast_at"] is None
|
||||
|
||||
|
||||
def test_f_commit_callback_updates_last_broadcast(mem_db, no_photon):
|
||||
"""After the dispatcher calls the callback, last_broadcast_* reflect
|
||||
the committed timestamp + the acres/containment of THIS broadcast."""
|
||||
env = _make_active_envelope(geocoder_city="Burley")
|
||||
wire, data = _run_handler_only(env, now=20_000)
|
||||
assert wire is not None
|
||||
|
||||
_commit(data, committed_at=20_005.0)
|
||||
|
||||
fr = mem_db.execute(
|
||||
"SELECT last_broadcast_at, last_broadcast_acres, last_broadcast_contained "
|
||||
"FROM fires WHERE irwin_id=?", (_IRWIN_A,)).fetchone()
|
||||
assert fr["last_broadcast_at"] == 20_005
|
||||
assert fr["last_broadcast_acres"] == 1847.0
|
||||
assert fr["last_broadcast_contained"] == 23
|
||||
|
||||
# Third pass: same IRWIN, no growth, no callback (cooldown applies).
|
||||
# Handler must return None this time because last_broadcast_at IS NOT NULL
|
||||
# and the change-detection gates report no change.
|
||||
env_same = _make_active_envelope(geocoder_city="Burley")
|
||||
wire3, _ = _run_handler_only(env_same, now=20_010)
|
||||
assert wire3 is None
|
||||
|
||||
|
||||
def test_g_callback_not_called_means_last_broadcast_stays_null(mem_db, no_photon):
|
||||
"""If dispatcher drops for any reason (grace, staleness, cooldown,
|
||||
dedup) the callback is not invoked -- last_broadcast_* stays NULL and
|
||||
the next successful broadcast emits 'New:' (not 'Update:'). This is
|
||||
the inverse of test_e from the persistence-row side."""
|
||||
env = _make_active_envelope(geocoder_city="Burley")
|
||||
wire, data = _run_handler_only(env, now=30_000)
|
||||
assert wire is not None
|
||||
# No _commit() call.
|
||||
fr = mem_db.execute(
|
||||
"SELECT last_broadcast_at FROM fires WHERE irwin_id=?",
|
||||
(_IRWIN_A,)).fetchone()
|
||||
assert fr["last_broadcast_at"] is None
|
||||
|
||||
|
||||
def test_h_no_audit_row_inserted_when_handler_skips_commit(mem_db, no_photon):
|
||||
"""The handler no longer writes mesh_broadcasts_out -- the dispatcher
|
||||
inserts it via `_broadcast_audit`. Until the dispatcher calls _commit,
|
||||
there should be zero rows in mesh_broadcasts_out even though fires
|
||||
has the new row."""
|
||||
env = _make_active_envelope(geocoder_city="Burley")
|
||||
wire, data = _run_handler_only(env, now=40_000)
|
||||
assert wire is not None
|
||||
n = mem_db.execute(
|
||||
"SELECT COUNT(*) AS n FROM mesh_broadcasts_out").fetchone()["n"]
|
||||
assert n == 0
|
||||
# The handler signalled the dispatcher SHOULD insert an audit row.
|
||||
audit = data["_broadcast_audit"]
|
||||
assert audit == {"table": "fires", "pk": _IRWIN_A}
|
||||
|
||||
|
||||
def test_h_handler_attaches_audit_descriptor_and_callback(mem_db, no_photon):
|
||||
"""Sanity: every active wire-string return must come with the two
|
||||
dispatcher hooks attached."""
|
||||
env = _make_active_envelope(geocoder_city="Burley", irwin_id=_IRWIN_B)
|
||||
data = {}
|
||||
wire = handle_wfigs(cn.normalize(env), env, env["subject"],
|
||||
data=data, now=50_000)
|
||||
assert wire is not None
|
||||
assert callable(data["_on_broadcast_committed"])
|
||||
assert data["_broadcast_audit"]["table"] == "fires"
|
||||
assert data["_broadcast_audit"]["pk"] == _IRWIN_B
|
||||
Loading…
Add table
Add a link
Reference in a new issue