Compare commits

...

9 commits

Author SHA1 Message Date
Matt Johnson
522458f194 fix: sync _TOWN_ANCHORS_SEED with the production town_anchors table
_TOWN_ANCHORS_SEED covered only 29 hand-picked towns, while the live
CT108 town_anchors table (GUI-curated since) has grown to 186. Any
fresh deploy (fresh volume, or the pre-v19 sqlite path) would seed
only the original 29 and silently lose anchor-resolution coverage for
the other 157 real, already-in-production towns.

Regenerated the dict from a live dump of CT108's town_anchors
(name, lat, lon, state; all 186 rows confirmed enabled=1), alphabetized,
same dict-of-dicts shape and column-aligned brace formatting as before
(alignment column widened to fit the longest name, "mountain home
afb"). seed_town_anchors() is unchanged and still INSERT OR IGNORE, so
it stays idempotent against the existing production rows.

Expanding the seed changes which town resolves as "nearest" for a few
existing test fixtures whose incident/work-zone coordinates are
genuinely closer to a newly-added real town (e.g. Oakley, Wilder) than
to the old 29-town subset's nearest match -- those goldens and
assertions are updated to the new (correct) nearest-town result.
test_api_post_add_town's probe town is renamed from the now-real
"Bellevue" to a fictitious "Testopolis" to avoid a duplicate-name 400.
2026-08-16 02:58:11 +00:00
Matt Johnson
3961f7ec04 fix: honour town_anchors.enabled in alert anchor-resolution queries
resolve_anchor() (notifications/formatters/_anchor.py) and
_location_anchor() (env/fire_render.py) selected all town_anchors rows
regardless of the enabled flag, so a disabled anchor could still be
used in outbound alert text -- only the dashboard/curation routes are
meant to see disabled rows. Add AND enabled = 1 to both queries.

No-op today: all 186 live town_anchors rows are enabled=1. Matters the
next time someone disables an anchor from the curation UI.

Also updates the traffic_last/0003.json wzdx golden literal in
test_incident_refactor.py and adds TestAnchorResolve::
test_disabled_anchor_excluded, which forces the Photon fallback to
miss and asserts a disabled-only DB row is not selected.
2026-08-16 02:58:00 +00:00
Matt Johnson
bdbc2afe7f MeshCore reconnect persistence: implement 0=unlimited max-reconnect-attempts sentinel
connect() now translates a configured meshcore_max_reconnect_attempts of 0
(or <=0) into an effectively-unbounded count before handing it to the
meshcore library's ConnectionManager, so its retry loop never
permanently exhausts. config.py's comment already documented \"0 =
unlimited\" but that sentinel was never actually implemented -- literal 0
meant zero attempts, and the shipped default of 5 (at ~1s/attempt) gave up
after ~5s with no external supervisor to retry again, leaving MeshCore dead
until a manual container restart. Also flips the repo default from 5 to 0
so fresh deploys get unlimited retries without extra config.

Proven via a 60s forced-outage auto-recovery test: the link recovers from
any-length vnode/radio outage instead of giving up after ~5s.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-08-16 01:24:47 +00:00
Matt Johnson
c5aa0e1f42 Fix event-loop starvation, MeshCore stability, config-page hardening
- mesh_data_store.py / env/store.py: make refresh() async, offload blocking
  polls via asyncio.to_thread/gather so 7 lockstep sources no longer starve
  the shared event loop.
- main.py: gather pollers concurrently + set_default_executor thread pool.
- Dockerfile / docker-compose.yml: healthcheck now curls the dashboard for a
  real liveness signal instead of a process-exists check.
- transport/meshcore_transport.py: MeshCore keepalive loop (get_time() every
  120s), reconnect re-arm (_post_reconnect_setup_async from
  _on_connect_event), and MC channel-name normalization
  (_resolve_mc_channel_idx strips a leading #).
- dashboard-frontend: MeshCoreConnection.tsx config-page hardening, new
  ErrorBoundary component, wired into App.tsx.
- tests: fix ~40 call sites broken by refresh() becoming async (
  test_generic_http.py, test_store_received_delta.py,
  test_store_wzdx_persist.py) by wrapping with asyncio.run(), matching this
  suite's existing convention for calling async code from sync test
  functions. Verified: all 40 tests pass.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-08-16 01:24:45 +00:00
Matt Johnson
3c281c96e2 fix(fire): honor tombstoned_at as a durable mute in the gate decider
fires.tombstoned_at (the same column the dashboard's active views
already filter on: /api/env/active WHERE tombstoned_at IS NULL) is now
checked in decide() before the row-missing/growth/cooldown branches,
so a tombstoned incident can never re-broadcast through any path.
Defensive on both the read (falls back to the pre-v12 column set if
tombstoned_at doesn't exist) and the lookup (missing key reads as "not
tombstoned" rather than raising) -- an un-migrated schema must not
crash the decider.
2026-08-16 01:20:18 +00:00
Matt Johnson
81ade1e323 fix(satpass): region-tag satpass events via observer_list
Satpass events carry no lat/lon/geometry, so CoverageFilter had no way
to region-tag them for region_routes matching. gate_consolidated_pass()
now attaches observer_list (comma-joined observer slugs, same shape
already computed for the audit column) into the event data, which
coverage_area.observer_region_names() reads. Defensive/fail-open: an
empty observer_list yields no region tag rather than raising.
2026-08-16 01:20:12 +00:00
Matt Johnson
738ffc71c3 fix(usgs_quake): stop dropping first-sighted quakes at the freshness gate
The USGS feed is a rolling PAST-DAY feed (2.5_day.geojson), so a
genuinely first-seen quake's age-at-ingest routinely exceeds the
generic per-toggle freshness_seconds (600s) window, silently dropping
it downstream of decide()/DB-insert (last_broadcast_at stayed NULL on
every sampled first-sighting row).

- dispatcher.py: earthquake_event gets its own adapter_config-backed
  freshness override (adapter_config.usgs_quake.freshness_seconds,
  default 3600s), scoped to the CATEGORY (not the "seismic" toggle,
  which also carries hydro stream_flood_warning/stream_high_water and
  must keep the generic per-toggle freshness).
- usgs_quake.py: to_event() no longer passes the adapter's fixed
  freshness in, deferring to the dispatcher override.
- adapter_config/defaults.py: adds the freshness_seconds field/default
  for usgs_quake.
2026-08-16 01:20:06 +00:00
Matt Johnson
50553b26f5 fix(roads511): distinguish full closures from partial restrictions
Adds explicit full-closure phrase matching (with partial-restriction
phrases as vetoes) for sub_type mapping in to_event(), so wording like
"All Shoulders Closed" or "Right Lane Closed" no longer gets broadcast
as a full road_closed event. Existing is_closure severity/summary logic
is untouched; the new is_full_closure_flag/_has_full_closure_language
helpers are additive and scoped to sub_type only.
2026-08-16 01:19:56 +00:00
Matt Johnson
c4706d3c92 fix(ipaws): route county-only alerts through evac-phase detection
Adds meshai.notifications.evac_phase.detect_phase, wired into the IPAWS
formatter/adapter, to classify READY/SET/GO evacuation phases from real
FEMA IPAWS alert headline/CMAMtext strings. Includes explicit
false-positive guards since a wrong GO detection would broadcast an
evacuation order that was never issued.
2026-08-16 01:19:46 +00:00
37 changed files with 2142 additions and 197 deletions

View file

@ -91,8 +91,8 @@ VOLUME ["/data"]
EXPOSE 8080
# Health check - verify bot process is alive via PID file
HEALTHCHECK --interval=30s --timeout=10s --start-period=10s --retries=3 \
CMD test -f /tmp/meshai.pid && kill -0 $(cat /tmp/meshai.pid) 2>/dev/null && [ "$(cat /tmp/meshai.link 2>/dev/null)" = up ] || exit 1
HEALTHCHECK --interval=30s --timeout=10s --start-period=240s --retries=3 \
CMD curl -f -s -o /dev/null http://localhost:8080/ || exit 1
# Entrypoint writes default config on first run, then starts the bot
ENTRYPOINT ["/app/docker-entrypoint.sh"]

View file

@ -18,12 +18,14 @@ import MeshCoreDangerZones from './pages/MeshCoreDangerZones'
import Coverage from './pages/Coverage'
import { ToastProvider } from './components/ToastProvider'
import { DirtyProvider } from './context/DirtyContext'
import ErrorBoundary from './components/ErrorBoundary'
function App() {
return (
<DirtyProvider>
<ToastProvider>
<Layout>
<ErrorBoundary>
<Routes>
{/* Core routes */}
<Route path="/" element={<Dashboard />} />
@ -66,6 +68,7 @@ function App() {
<Route path="/meshcore/companion" element={<Navigate to="/meshcore/contacts" replace />} />
<Route path="/meshcore/danger-zones" element={<MeshCoreDangerZones />} />
</Routes>
</ErrorBoundary>
</Layout>
</ToastProvider>
</DirtyProvider>

View file

@ -0,0 +1,49 @@
// App-wide render-error safety net. Wraps <Routes> in App.tsx so an
// unhandled error thrown while rendering any page degrades to a recoverable
// "Something went wrong" card instead of a blank white screen.
import { Component, type ErrorInfo, type ReactNode } from 'react'
import { AlertTriangle } from 'lucide-react'
interface Props {
children: ReactNode
}
interface State {
hasError: boolean
error: Error | null
}
export default class ErrorBoundary extends Component<Props, State> {
state: State = { hasError: false, error: null }
static getDerivedStateFromError(error: Error): State {
return { hasError: true, error }
}
componentDidCatch(error: Error, errorInfo: ErrorInfo) {
console.error('MeshAI dashboard render error:', error, errorInfo)
}
render() {
if (this.state.hasError) {
return (
<div className="flex items-center justify-center h-64">
<div className="bg-bg-card border border-red-500/20 rounded p-6 max-w-md w-full text-center space-y-3">
<AlertTriangle className="mx-auto text-red-400" size={28} />
<div className="text-slate-200 font-medium">Something went wrong</div>
<div className="text-xs text-slate-500">
{this.state.error?.message ?? 'An unexpected error occurred while rendering this page.'}
</div>
<button
onClick={() => window.location.reload()}
className="px-4 py-2 bg-accent hover:bg-accent/80 rounded text-white text-sm transition-colors"
>
Reload
</button>
</div>
</div>
)
}
return this.props.children
}
}

View file

@ -1,6 +1,6 @@
import { useState, useEffect, useCallback } from 'react'
import { Link } from 'react-router-dom'
import { Save, RotateCcw, RefreshCw, Check, ChevronRight, Trash2, Eye, EyeOff, Copy } from 'lucide-react'
import { Save, RotateCcw, RefreshCw, Check, ChevronRight, Trash2, Eye, EyeOff, Copy, X } from 'lucide-react'
import { TextInput, NumberInput, Toggle, ListInput, SelectInput } from './Config'
import SerialPortPicker from '@/components/SerialPortPicker'
import { notifyRestartRequired } from '@/components/RestartBanner'
@ -58,6 +58,9 @@ export default function MeshCoreConnection() {
const [error, setError] = useState<string | null>(null)
const [success, setSuccess] = useState<string | null>(null)
const [hasChanges, setHasChanges] = useState(false)
// Set when one or both config fetches fail; the form still renders (using
// safe defaults for whatever didn't load) instead of blanking the page.
const [loadError, setLoadError] = useState<string | null>(null)
// Test send state
const [channelsActive, setChannelsActive] = useState(false)
@ -81,22 +84,41 @@ export default function MeshCoreConnection() {
const fetchConfig = useCallback(async () => {
setLoading(true)
try {
const [data, mcCtx] = await Promise.all([
apiFetchConfig('connection') as Promise<ConnectionConfig>,
apiFetchConfig('meshcore_context') as Promise<MeshcoreContextCfg>,
])
setConfig(data)
setOriginalConfig(JSON.parse(JSON.stringify(data)))
setMcContext(mcCtx)
setOriginalMcContext(JSON.parse(JSON.stringify(mcCtx)))
setHasChanges(false)
setError(null)
} catch (err) {
setError(err instanceof Error ? err.message : 'Unknown error')
} finally {
setLoading(false)
// Promise.allSettled so one section failing to load can't blank the
// whole page — each section is set independently from its own result,
// and any failure(s) are surfaced as a dismissible banner alongside the
// still-usable form (see the `!config` dead-end this replaces).
const [connResult, mcResult] = await Promise.allSettled([
apiFetchConfig('connection') as Promise<ConnectionConfig>,
apiFetchConfig('meshcore_context') as Promise<MeshcoreContextCfg>,
])
const errors: string[] = []
if (connResult.status === 'fulfilled') {
setConfig(connResult.value)
setOriginalConfig(JSON.parse(JSON.stringify(connResult.value)))
} else {
errors.push(`connection (${connResult.reason instanceof Error ? connResult.reason.message : String(connResult.reason)})`)
// Fall back to an empty object (never null) so the form below always
// has something to render safe defaults from, and so the dirty-check
// diff below still has a baseline to compare edits against.
setConfig((c) => c ?? {})
setOriginalConfig((c) => c ?? {})
}
if (mcResult.status === 'fulfilled') {
setMcContext(mcResult.value)
setOriginalMcContext(JSON.parse(JSON.stringify(mcResult.value)))
} else {
errors.push(`bot behavior (${mcResult.reason instanceof Error ? mcResult.reason.message : String(mcResult.reason)})`)
// Leave mcContext null — the "Bot behavior" card only renders when it
// is present, so a null value just hides that card cleanly.
}
setLoadError(errors.length ? `Couldn't load ${errors.join(' and ')}. Showing defaults — edits below are still safe to make and save.` : null)
setHasChanges(false)
setLoading(false)
}, [])
useEffect(() => {
@ -207,10 +229,13 @@ export default function MeshCoreConnection() {
}
useEffect(() => {
if (config && originalConfig && mcContext && originalMcContext) {
// mcContext may be null (its fetch failed) — don't let that block
// detecting changes to the connection fields, which always load or
// fall back to an empty-object baseline in fetchConfig.
if (config && originalConfig) {
const changed =
JSON.stringify(config) !== JSON.stringify(originalConfig) ||
JSON.stringify(mcContext) !== JSON.stringify(originalMcContext)
(!!mcContext && !!originalMcContext && JSON.stringify(mcContext) !== JSON.stringify(originalMcContext))
setHasChanges(changed)
}
}, [config, originalConfig, mcContext, originalMcContext])
@ -221,22 +246,24 @@ export default function MeshCoreConnection() {
}, [hasChanges, setDirty])
const upd = (patch: Partial<ConnectionConfig>) =>
setConfig((c) => (c ? { ...c, ...patch } : c))
// Build off an empty object rather than bailing when config is still
// null (e.g. mid-retry) — edits should never be silently dropped.
setConfig((c) => ({ ...(c ?? {}), ...patch }))
const saveConfig = async () => {
if (!config || !mcContext) return
if (!config) return
setSaving(true)
setError(null)
setSuccess(null)
try {
// PUT the whole objects so sibling fields (Meshtastic connection fields,
// any other meshcore_context keys) are preserved.
const results = await Promise.all([
apiUpdateConfig('connection', config),
apiUpdateConfig('meshcore_context', mcContext),
])
// any other meshcore_context keys) are preserved. Only PUT
// meshcore_context if it actually loaded — its fetch may have failed.
const puts: Promise<{ restart_required?: boolean }>[] = [apiUpdateConfig('connection', config)]
if (mcContext) puts.push(apiUpdateConfig('meshcore_context', mcContext))
const results = await Promise.all(puts)
setOriginalConfig(JSON.parse(JSON.stringify(config)))
setOriginalMcContext(JSON.parse(JSON.stringify(mcContext)))
if (mcContext) setOriginalMcContext(JSON.parse(JSON.stringify(mcContext)))
setHasChanges(false)
setDirty(false)
setSuccess('MeshCore connection saved successfully')
@ -276,13 +303,10 @@ export default function MeshCoreConnection() {
)
}
if (!config) {
return (
<div className="flex items-center justify-center h-64">
<div className="text-red-400">Failed to load connection config</div>
</div>
)
}
// Always render the form below, even if the fetch failed — `cfg` supplies
// safe defaults so the connection-type/host/port fields and Save/Discard
// stay usable. The failure itself is surfaced by the loadError banner.
const cfg: ConnectionConfig = config ?? {}
return (
<div className="max-w-2xl mx-auto space-y-6">
@ -320,6 +344,30 @@ export default function MeshCoreConnection() {
</div>
</div>
{/* Load-error banner dismissible, with its own Retry so the page
never dead-ends when a config section fails to fetch. The form
below stays fully editable regardless. */}
{loadError && (
<div className="p-3 text-sm bg-red-500/10 text-red-400 border border-red-500/20 flex items-start gap-3">
<div className="flex-1">{loadError}</div>
<button
onClick={fetchConfig}
className="flex items-center gap-1 px-2 py-1 bg-red-500/20 hover:bg-red-500/30 rounded text-red-300 text-xs shrink-0"
>
<RefreshCw size={12} />
Retry
</button>
<button
onClick={() => setLoadError(null)}
title="Dismiss"
aria-label="Dismiss load error"
className="text-red-400 hover:text-red-200 shrink-0"
>
<X size={14} />
</button>
</div>
)}
{/* Status messages */}
{error && (
<div className="p-3 text-sm bg-red-500/10 text-red-400 border border-red-500/20">{error}</div>
@ -341,7 +389,7 @@ export default function MeshCoreConnection() {
</p>
<SelectInput
label="Connection Type"
value={config.meshcore_conn_type ?? 'tcp'}
value={cfg.meshcore_conn_type ?? 'tcp'}
onChange={(v) => upd({ meshcore_conn_type: v })}
options={[
{ value: 'tcp', label: 'TCP (companion)' },
@ -350,11 +398,11 @@ export default function MeshCoreConnection() {
]}
helper="TCP for a companion frame server, Serial for a USB node, BLE for Bluetooth"
/>
{(config.meshcore_conn_type ?? 'tcp') === 'tcp' && (
{(cfg.meshcore_conn_type ?? 'tcp') === 'tcp' && (
<div className="grid grid-cols-2 gap-4">
<TextInput
label="MeshCore Host"
value={config.meshcore_host ?? ''}
value={cfg.meshcore_host ?? ''}
onChange={(v) => upd({ meshcore_host: v })}
placeholder="192.168.1.100"
helper="IP or hostname of the companion frame server"
@ -362,7 +410,7 @@ export default function MeshCoreConnection() {
/>
<NumberInput
label="MeshCore Port"
value={config.meshcore_port ?? 5525}
value={cfg.meshcore_port ?? 5525}
onChange={(v) => upd({ meshcore_port: v })}
min={1}
max={65535}
@ -370,27 +418,27 @@ export default function MeshCoreConnection() {
/>
</div>
)}
{(config.meshcore_conn_type ?? 'tcp') === 'serial' && (
{(cfg.meshcore_conn_type ?? 'tcp') === 'serial' && (
<>
<SerialPortPicker
label="MeshCore Serial Port"
value={config.meshcore_serial_port ?? ''}
value={cfg.meshcore_serial_port ?? ''}
onChange={(v) => upd({ meshcore_serial_port: v })}
helper="USB-attached MeshCore node — Detect fills a stable by-id path"
/>
<NumberInput
label="Baud Rate"
value={config.meshcore_baud ?? 115200}
value={cfg.meshcore_baud ?? 115200}
onChange={(v) => upd({ meshcore_baud: v })}
min={1200}
helper="Serial baud rate (default 115200)"
/>
</>
)}
{(config.meshcore_conn_type ?? 'tcp') === 'ble' && (
{(cfg.meshcore_conn_type ?? 'tcp') === 'ble' && (
<TextInput
label="BLE Address"
value={config.meshcore_ble_address ?? ''}
value={cfg.meshcore_ble_address ?? ''}
onChange={(v) => upd({ meshcore_ble_address: v })}
placeholder="AA:BB:CC:DD:EE:FF"
helper="Leave blank to scan/pair the first available device"
@ -406,7 +454,7 @@ export default function MeshCoreConnection() {
</div>
<Toggle
label="Auto-add contacts (AIDA adds any node it hears — required to DM anyone)"
checked={config.meshcore_auto_add_contacts ?? true}
checked={cfg.meshcore_auto_add_contacts ?? true}
onChange={(v) => upd({ meshcore_auto_add_contacts: v })}
helper="Enables firmware CMD 58 (set_autoadd_config) at connect so AIDA automatically adds every node it hears an advert from as a contact, enabling DM send/decrypt without manual contact exchange"
/>
@ -418,13 +466,13 @@ export default function MeshCoreConnection() {
<div className="mt-4 space-y-4 pl-6 border-l border-[#1e2a3a]">
<Toggle
label="Auto-reconnect (MeshCore)"
checked={config.meshcore_auto_reconnect ?? true}
checked={cfg.meshcore_auto_reconnect ?? true}
onChange={(v) => upd({ meshcore_auto_reconnect: v })}
helper="Automatically reconnect to the MeshCore companion if the link drops"
/>
<NumberInput
label="Max Reconnect Attempts"
value={config.meshcore_max_reconnect_attempts ?? 5}
value={cfg.meshcore_max_reconnect_attempts ?? 5}
onChange={(v) => upd({ meshcore_max_reconnect_attempts: v })}
min={0}
helper="Maximum reconnect attempts before giving up (0 = unlimited)"

View file

@ -75,11 +75,11 @@ services:
memory: 64M
healthcheck:
test: ["CMD-SHELL", "test -f /tmp/meshai.pid && kill -0 $(cat /tmp/meshai.pid) 2>/dev/null && [ \"$(cat /tmp/meshai.link 2>/dev/null)\" = up ] || exit 1"]
test: ["CMD-SHELL", "curl -f -s -o /dev/null http://localhost:8080/ || exit 1"]
interval: 30s
timeout: 10s
retries: 3
start_period: 15s
start_period: 240s
logging:
driver: "json-file"

View file

@ -85,7 +85,8 @@ REGISTRY: dict[tuple[str, str], dict[str, Any]] = {
},
# =================================================================
# USGS_QUAKE -- 6 settings (regional geography + 3 mag floors + PAGER set)
# USGS_QUAKE -- 7 settings (regional geography + 3 mag floors + PAGER set
# + freshness gate)
# =================================================================
("usgs_quake", "regional_centroid"): {
"default": [44.36, -114.61], # quake_handler.py:36-37 (Idaho centroid)
@ -117,6 +118,11 @@ REGISTRY: dict[tuple[str, str], dict[str, Any]] = {
"type": "float",
"description": "Magnitude floor for the visual escalation emoji.",
},
("usgs_quake", "freshness_seconds"): {
"default": 3600,
"type": "int",
"description": "Staleness gate for earthquake_event broadcasts (age = now - quake origin time), read by the dispatcher instead of the generic per-toggle freshness. The upstream feed is a rolling PAST-DAY feed, so quakes are routinely detected 10min-20h after origin; 1 hour keeps quakes broadcast-worthy (still situationally relevant) while excluding the day-feed's stale backlog. 0 = disabled (not recommended -- would readmit day-old quakes).",
},
# =================================================================
# SWPC -- 3 settings (three storm-tier broadcast floors)

View file

@ -51,7 +51,7 @@ class ConnectionConfig:
meshcore_host: str = "" # pyMC companion frame server host
meshcore_port: int = 5050 # pyMC companion frame server port
meshcore_auto_reconnect: bool = True # enable meshcore lib auto-reconnect
meshcore_max_reconnect_attempts: int = 5 # max reconnect attempts (0 = unlimited)
meshcore_max_reconnect_attempts: int = 0 # max reconnect attempts (0 = unlimited)
meshcore_advert_interval_seconds: int = 86400 # periodic self-advert interval, 24h (0 = disabled)
# MeshCore connection type: tcp | serial | ble (default tcp for back-compat)
meshcore_conn_type: str = "tcp"

View file

@ -248,7 +248,7 @@ def _location_anchor(n: dict) -> str:
try:
from meshai.persistence import get_db
rows = get_db().execute(
"SELECT name, lat, lon FROM town_anchors WHERE lat IS NOT NULL AND lon IS NOT NULL"
"SELECT name, lat, lon FROM town_anchors WHERE lat IS NOT NULL AND lon IS NOT NULL AND enabled = 1"
).fetchall()
best = None
best_d = float("inf")

View file

@ -104,6 +104,13 @@ def _norm_fips(value) -> str:
class IPAWSAlertsAdapter:
"""FEMA IPAWS-OPEN EAS civil alerts — two-stage CAP poller."""
# Stage-2 retry suppression. 401/403/404/410 are treated as durable (the
# alert body will not become fetchable), so we wait out the feed's rolling
# window; everything else (timeout, 5xx, 429, connection) is transient.
_STAGE2_FORBIDDEN_COOLDOWN = 21600 # 6h
_STAGE2_TRANSIENT_COOLDOWN = 300 # 5m
_STAGE2_FORBIDDEN_CODES = (401, 403, 404, 410)
def __init__(self, config: "IPAWSConfig", coverage: dict = None):
self._base_url = _cfg_str(config, "base_url", DEFAULT_BASE_URL).rstrip("/")
self._user_agent = getattr(config, "user_agent", "") or "meshai-ipaws/1.0"
@ -133,6 +140,11 @@ class IPAWSAlertsAdapter:
self._last_error = None
self._backoff_until = 0.0
self._is_loaded = False
# Negative cache: stage-2 CAP URL -> epoch after which a retry is
# allowed. Stops re-hammering FEMA for an alert listed in the feed whose
# detail endpoint keeps failing (e.g. a durable 403 on a COG-restricted
# alert). Pruned to the current feed each pass so it can't grow unbounded.
self._stage2_cooldown = {}
# ── Polling ──────────────────────────────────────────────────────────────
@ -205,6 +217,8 @@ class IPAWSAlertsAdapter:
self._consecutive_errors += 1
return False
now = time.time()
seen_urls = set()
new_events = []
for entry in feed.findall(f"{{{_ATOM_NS}}}entry"):
statefips = None
@ -221,18 +235,43 @@ class IPAWSAlertsAdapter:
cap_url = self._stage2_url(href)
if not cap_url:
continue
seen_urls.add(cap_url)
# ── Negative cache: skip URLs still within their failure cooldown ─
retry_after = self._stage2_cooldown.get(cap_url)
if retry_after is not None and now < retry_after:
continue
# ── Stage 2: full CAP document ───────────────────────────────────
try:
cap_raw = self._get(cap_url)
except Exception as e: # noqa: BLE001
logger.debug("IPAWS stage-2 fetch failed for %s: %s", cap_url, e)
except HTTPError as e:
cooldown = (self._STAGE2_FORBIDDEN_COOLDOWN
if e.code in self._STAGE2_FORBIDDEN_CODES
else self._STAGE2_TRANSIENT_COOLDOWN)
self._stage2_cooldown[cap_url] = now + cooldown
logger.debug("IPAWS stage-2 fetch failed for %s: HTTP %s "
"(cooldown %ds)", cap_url, e.code, cooldown)
continue
except Exception as e: # noqa: BLE001
self._stage2_cooldown[cap_url] = now + self._STAGE2_TRANSIENT_COOLDOWN
logger.debug("IPAWS stage-2 fetch failed for %s: %s (cooldown %ds)",
cap_url, e, self._STAGE2_TRANSIENT_COOLDOWN)
continue
# Success — clear any stale cooldown for this URL.
self._stage2_cooldown.pop(cap_url, None)
parsed = self._parse_cap(cap_raw, statefips)
if parsed:
new_events.append(parsed)
# Prune the negative cache to URLs still present in the feed, so it
# tracks the rolling window and never grows without bound.
self._stage2_cooldown = {
u: t for u, t in self._stage2_cooldown.items() if u in seen_urls
}
# Change detection on the active identifier set.
old_ids = {e["event_id"] for e in self._events}
new_ids = {e["event_id"] for e in new_events}
@ -297,6 +336,30 @@ class IPAWSAlertsAdapter:
certainty = _t(info, "certainty")
headline = _t(info, "headline")
description = _t(info, "description")
instruction = _t(info, "instruction")
# ── <parameter> blocks: valueName -> value ────────────────────────────
# Civil alerts carry their real public-facing text here (CMAMtext /
# CMAMlongtext are the agency's own purpose-written WEA copy) — this was
# previously discarded entirely. A repeated valueName (e.g. multiple
# BLOCKCHANNEL entries) becomes a list; defensive against malformed/
# missing valueName or value children (skipped, never raises).
parameters: dict = {}
for param in info.findall(f"{{{_CAP_NS}}}parameter"):
vn = param.find(f"{{{_CAP_NS}}}valueName")
vv = param.find(f"{{{_CAP_NS}}}value")
name = vn.text.strip() if vn is not None and vn.text else ""
if not name:
continue
value = vv.text.strip() if vv is not None and vv.text else ""
if name in parameters:
existing = parameters[name]
if isinstance(existing, list):
existing.append(value)
else:
parameters[name] = [existing, value]
else:
parameters[name] = value
# eventCode SAME value
same_value = ""
@ -377,6 +440,8 @@ class IPAWSAlertsAdapter:
"msgType": msg_type,
"headline": headline,
"description": description,
"instruction": instruction,
"parameters": parameters,
"same_code": same_value,
"area_desc": area_desc,
"area_same_codes": area_same_codes,
@ -491,7 +556,8 @@ class IPAWSAlertsAdapter:
"area_desc": area_desc,
"geocoder": {"city": None, "county": area_desc, "state": None},
"description": raw.get("description", ""),
"parameters": {},
"instruction": raw.get("instruction", ""),
"parameters": raw.get("parameters") or {},
"msgType": raw.get("msgType", "Alert"),
"references": [],
"category": category,

View file

@ -21,6 +21,53 @@ if TYPE_CHECKING:
logger = logging.getLogger(__name__)
# Explicit full-road-closure language, matched case-insensitively. Used
# ONLY by to_event()'s sub_type mapping (branch a) to decide "road_closed"
# vs. "road_works" — distinct from (and stricter than) the adapter's
# existing `is_closure` property, which loosely matches any "closed"
# substring and therefore also flags partial-restriction wording (e.g.
# "All Shoulders Closed") that the upstream owner explicitly does not want
# broadcast as a closure.
_FULL_CLOSURE_PHRASES = (
"all lanes closed",
"road closed",
"highway closed",
"fully closed",
"closed in both directions",
)
# Partial-restriction phrases that must NEVER be counted as a full closure.
# Checked first so one of these vetoes a match even if closure wording
# happens to appear elsewhere in the same description — e.g. "all
# shoulders closed" must not fall through to a naive "closed" test, and
# "one/right/left lane closed" must not be confused with "all lanes
# closed" (note the singular "lane closed" vs. plural "lanes closed").
_PARTIAL_RESTRICTION_PHRASES = (
"shoulder closed",
"shoulders closed",
"lane closed",
"ramp closed",
"alternating",
"lane restriction",
)
def _has_full_closure_language(description: str) -> bool:
"""True if `description` contains explicit full-closure phrasing.
Case-insensitive. Partial-restriction phrases (shoulder/lane/ramp
closures, alternating traffic) are checked first and veto a match, so
they never register as a full closure regardless of other wording
present.
"""
text = (description or "").lower()
if any(phrase in text for phrase in _PARTIAL_RESTRICTION_PHRASES):
return False
return any(phrase in text for phrase in _FULL_CLOSURE_PHRASES)
class Roads511Adapter:
"""511 road conditions polling adapter."""
@ -284,12 +331,21 @@ class Roads511Adapter:
lat = loc.get("latitude") or loc.get("lat")
lon = loc.get("longitude") or loc.get("lon") or loc.get("lng")
# Check closure status
is_closure = (
# Raw upstream full-closure signal, captured on its own (no
# description heuristic mixed in) so to_event() can build a
# stricter full-closure determination for sub_type mapping
# without touching is_closure below, which severity/summary
# logic elsewhere already depends on.
is_full_closure_flag = bool(
item.get("IsFullClosure") or
item.get("is_full_closure") or
item.get("fullClosure") or
item.get("closed") or
item.get("closed")
)
# Check closure status
is_closure = (
is_full_closure_flag or
"closure" in str(event_type).lower() or
"closed" in str(description).lower()
)
@ -323,6 +379,19 @@ class Roads511Adapter:
None
)
# Granular ITD 511 v2 fields (roadwork vs. crash vs. closure
# discrimination). Defensive: absent on other states' 511 feeds.
event_sub_type = (
item.get("EventSubType") or
item.get("event_sub_type") or
""
)
cause = (
item.get("Cause") or
item.get("cause") or
None
)
# Default 6 hour TTL, refreshed every tick
expires = now + 21600
@ -346,7 +415,10 @@ class Roads511Adapter:
"properties": {
"roadway": roadway,
"is_closure": bool(is_closure),
"is_full_closure_flag": is_full_closure_flag,
"last_updated": last_updated,
"event_sub_type": event_sub_type,
"cause": cause,
},
}
@ -411,10 +483,49 @@ class Roads511Adapter:
# NOTE: Central-era rows used source 'itd_511' with 'idaho_511:event:*'
# ids — a different keyspace the pre-seed intentionally does not cover.
_external_id = evt.get("external_id") or event_id
# sub_type mapping: distinguish routine roadwork (suppressed by
# gating/incident.py's work-zone rule) from genuine news (crashes,
# closures, hazards). A construction-caused FULL closure is still
# real news and must broadcast, so a full closure wins regardless
# of EventType. Precedence:
# 1. full closure (raw flag OR explicit closure wording,
# see _has_full_closure_language) -> road_closed (always)
# 2. EventType roadwork /
# EventSubType has "construction" -> road_works (suppressed)
# 3. EventType accidentsAndIncidents -> incident
# 4. EventType closures -> road_closed
# 5. else (specialEvents, unknown) -> incident (fail open)
#
# NOTE: branch 1 deliberately does NOT reuse `_is_closure` above.
# `_is_closure` is a loose OR-chain (also used for severity and
# the summary text) that matches any "closed" substring in the
# description, including partial-restriction wording like "All
# Shoulders Closed" — routine construction the owner explicitly
# does not want broadcast. `_full_closure_for_mapping` is a
# stricter, mapping-only determination.
_event_type_val = str(evt.get("event_type") or "").strip().lower()
_event_sub_type_val = str(props.get("event_sub_type") or "").strip().lower()
_cause = props.get("cause")
_full_closure_for_mapping = bool(
props.get("is_full_closure_flag")
) or _has_full_closure_language(_desc)
if _full_closure_for_mapping:
_sub_type = "road_closed"
elif _event_type_val == "roadwork" or "construction" in _event_sub_type_val:
_sub_type = "road_works"
elif _event_type_val == "accidentsandincidents":
_sub_type = "incident"
elif _event_type_val == "closures":
_sub_type = "road_closed"
else:
_sub_type = "incident"
canonical_data = {
"external_id": _external_id, # 511_{itd_id}; enables durable pre-seed
"source": "511",
"sub_type": "road_closed" if _is_closure else "incident",
"sub_type": _sub_type,
"road": _roadway or None,
"direction": None, # not structured in native 511 feed
"from_loc": None,
@ -423,7 +534,7 @@ class Roads511Adapter:
"mile_end": None,
"mile_marker": None,
"lanes_affected": None,
"cause": None,
"cause": _cause,
"comment": _desc[:200] if _desc else title,
"impact": "all lanes closed" if _is_closure else None,
"county": None,

View file

@ -373,7 +373,20 @@ def gate_consolidated_pass(consolidated: dict, *,
# Prepare data dict with callbacks
severity_word = _map_severity(max_el)
data = {"_meshai_precomposed": True, "_severity_override": severity_word}
data = {
"_meshai_precomposed": True,
"_severity_override": severity_word,
# Region tagging: coverage_area.observer_region_names() reads
# event.data["observer_list"] as a comma-joined string of observer
# slugs (the same shape already computed above for the audit
# column). Satpass events carry no lat/lon/geometry, so this is the
# ONLY path that lets CoverageFilter region-tag a satpass event for
# region_routes matching. `observer_list` is already defensive
# (falls back to entry_obs, or "") so this never raises; an empty
# string is fail-open in observer_region_names() (-> no region tag,
# not an exception).
"observer_list": observer_list,
}
_attach_commit(data, event_id=consolidated_id, event_log_row_id=None)
return wire, data

View file

@ -1,5 +1,6 @@
"""Environmental data store with tick-based adapter polling."""
import asyncio
import hashlib
import json
import logging
@ -438,20 +439,38 @@ class EnvironmentalStore:
from meshai import coverage as _cov
return _cov.resolve_adapter_coverage(adapter, self._coverage_bbox, "native")
def refresh(self) -> bool:
async def refresh(self) -> bool:
"""Called every second from main loop. Ticks each adapter.
Adapter tick() calls (blocking network I/O) run concurrently in
worker threads and are AWAITED to completion before ingest, so the
event loop stays responsive while fetches are in flight. Ingest
(DB/EventBus work) then runs on the loop thread once all ticks are
done, exactly as before, so no thread ever overlaps ingest.
Returns:
True if any data changed
"""
changed = False
for name, adapter in self._adapters.items():
try:
if adapter.tick():
changed = True
adapters = list(self._adapters.items())
if not adapters:
self._purge_expired()
return changed
results = await asyncio.gather(
*(asyncio.to_thread(adapter.tick) for _, adapter in adapters),
return_exceptions=True,
)
for (name, adapter), result in zip(adapters, results):
if isinstance(result, Exception):
logger.warning("Env adapter %s error: %s", name, result)
continue
if result:
changed = True
try:
self._ingest(name, adapter)
except Exception as e:
logger.warning("Env adapter %s error: %s", name, e)
except Exception as e:
logger.warning("Env adapter %s error: %s", name, e)
self._purge_expired()
return changed

View file

@ -246,7 +246,14 @@ class USGSQuakeAdapter:
expires=evt.get("expires"),
lat=lat,
lon=lon,
region=evt.get("region"),
# Deliberately NOT passing region=evt.get("region") here: the
# raw dict's region is a fixed config default (self._region,
# "magic_valley"), which doesn't match any region_routes cell
# key. Event.region is documented as "set by region tagger"
# (notifications/events.py) -- leaving it unset lets
# CoverageFilter's geometry-based tagging (event.lat/lon vs
# the named coverage areas) populate the real region name
# from the quake's actual coordinates instead.
group_key=event_id,
inhibit_keys=[event_id],
data=canonical_data,

View file

@ -2,6 +2,7 @@
import argparse
import asyncio
import concurrent.futures
import logging
import os
import signal
@ -146,13 +147,46 @@ class MeshAI:
while self._running:
await asyncio.sleep(1)
# Periodic MeshMonitor refresh
if self.meshmonitor_sync:
self.meshmonitor_sync.maybe_refresh()
# Periodic data store refresh and health computation
# Run the mesh/env/meshmonitor pollers concurrently so blocking
# network I/O (tick() fetches) never starves this loop — and
# therefore never starves the dashboard, which shares this same
# asyncio loop. Each refresh() internally awaits its own due
# ticks in worker threads; meshmonitor_sync.maybe_refresh is
# synchronous, so it is offloaded to a thread here directly.
# We await the WHOLE cycle before the next iteration, so no
# tick() thread ever overlaps the next cycle's bookkeeping.
_refresh_tasks = {}
if self.data_store:
_refresh_tasks['data'] = self.data_store.refresh()
if self.env_store:
_refresh_tasks['env'] = self.env_store.refresh()
if self.meshmonitor_sync:
_refresh_tasks['mm'] = asyncio.to_thread(self.meshmonitor_sync.maybe_refresh)
if _refresh_tasks:
_refresh_results = dict(zip(
_refresh_tasks.keys(),
await asyncio.gather(*_refresh_tasks.values(), return_exceptions=True),
))
else:
_refresh_results = {}
refreshed = _refresh_results.get('data')
if isinstance(refreshed, Exception):
logger.warning("Data store refresh error: %s", refreshed)
refreshed = False
env_changed = _refresh_results.get('env')
if isinstance(env_changed, Exception):
logger.debug("Env refresh error: %s", env_changed)
env_changed = False
_mm_result = _refresh_results.get('mm')
if isinstance(_mm_result, Exception):
logger.warning("MeshMonitor sync refresh error: %s", _mm_result)
# Periodic data store health computation
if self.data_store:
refreshed = self.data_store.refresh()
# Recompute health after refresh
if refreshed and self.health_engine:
self.health_engine.compute(self.data_store)
@ -202,10 +236,9 @@ class MeshAI:
except Exception:
pass
# Environmental feed refresh
# Environmental feed alerting/broadcast (refresh already ran above)
if self.env_store:
try:
env_changed = self.env_store.refresh()
if env_changed and self.alert_engine:
env_alerts = self.alert_engine.check_environmental(self.env_store)
if env_alerts:
@ -930,6 +963,14 @@ def main() -> None:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
# Size the default executor generously: mesh sources + env adapters
# (~7 sources, ~15 env adapters) now fetch concurrently via
# asyncio.to_thread() every tick, so they need thread headroom to avoid
# queuing behind each other on the default executor's small pool.
loop.set_default_executor(
concurrent.futures.ThreadPoolExecutor(max_workers=24, thread_name_prefix="meshai-io")
)
def signal_handler(sig, frame):
logger.info(f"Received signal {sig}")
loop.create_task(bot.stop())

View file

@ -6,6 +6,7 @@ This module replaces mesh_sources.py with a clean three-layer architecture:
- Layer 3: Consumers read unified model (no field guessing)
"""
import asyncio
import json
import logging
import sqlite3
@ -441,12 +442,16 @@ class MeshDataStore:
if stale_nums:
logger.info(f"Purged {len(stale_nums)} stale nodes (not heard in {STALE_NODE_THRESHOLD_DAYS} days)")
def refresh(self) -> bool:
async def refresh(self) -> bool:
"""Tick-based refresh. Called every second from the main loop.
Delegates to source tick() for sources that support it.
Only does a full rebuild when nodes/edges/topology change.
Only does a lightweight update when only packets change.
Delegates to source tick() for sources that support it. Due sources'
tick() calls (blocking network I/O) run concurrently in worker
threads and are AWAITED to completion before any bookkeeping, so the
event loop (and therefore the dashboard) stays responsive while
fetches are in flight. Only does a full rebuild when nodes/edges/
topology change. Only does a lightweight update when only packets
change.
Returns:
True if any data changed
@ -456,26 +461,38 @@ class MeshDataStore:
needs_rebuild = False
needs_packet_update = False
due: list[tuple[str, object]] = []
for name, source in self._sources.items():
# Check if this source supports tick-based polling
if hasattr(source, 'tick') and hasattr(source, '_tick_interval'):
if now - source._last_tick >= source._tick_interval:
endpoint = source.tick()
if endpoint:
any_changed = True
# Major changes require full rebuild
if endpoint in ("nodes", "edges", "traceroutes", "topology", "telemetry"):
needs_rebuild = True
# Packet-only changes are lightweight
elif endpoint in ("packets",):
needs_packet_update = True
# stats, counts, channels, solar, network just update cached data
due.append((name, source))
else:
# Legacy fallback for sources without tick support
if source.maybe_refresh():
any_changed = True
needs_rebuild = True
if due:
results = await asyncio.gather(
*(asyncio.to_thread(source.tick) for _, source in due),
return_exceptions=True,
)
for (name, source), result in zip(due, results):
if isinstance(result, Exception):
logger.warning(f"Source {name} tick failed: {result}")
continue
endpoint = result
if endpoint:
any_changed = True
# Major changes require full rebuild
if endpoint in ("nodes", "edges", "traceroutes", "topology", "telemetry"):
needs_rebuild = True
# Packet-only changes are lightweight
elif endpoint in ("packets",):
needs_packet_update = True
# stats, counts, channels, solar, network just update cached data
if needs_rebuild:
self._rebuild()
self._purge_stale_nodes()

View file

@ -0,0 +1,95 @@
"""Idaho READY / SET / GO evacuation-phase detection from free-text CAP content.
Real-world FEMA IPAWS civil alerts do NOT carry a machine-readable phase
CAP ``<responseType>`` does not exist in the wild. The phase (READY / SET /
GO) instead shows up as free text inside ``headline``, ``description``, and
the ``CMAMtext``/``CMAMlongtext`` ``<parameter>`` values, phrased however the
issuing agency happened to write it ("Level 3 GO NOW", "Set to GO",
"Evacuation Warning", ...). This module scans that free text for the phrases
agencies actually use and returns the phase, or ``None`` when nothing
matches callers must never guess a phase, since a false GO would broadcast
an evacuation order that was never issued.
"""
from __future__ import annotations
import re
# ── Strong multi-word phrases (checked case-insensitively) ───────────────────
# ANY match sets that phase's hit flag. Order within a list is irrelevant —
# the final result is decided purely by GO > SET > READY precedence below,
# never by which phrase or text argument matched first.
_GO_PHRASES = [
r"\bGO\s+NOW\b",
r"\bLEVEL\s+3\s+GO\b",
r"\bLEVEL\s+3\b",
r"\bLEVEL\s+III\b",
r"\bGO\s+EVACUATION\b",
r"\bIMMEDIATE\s+EVACUATION\b",
r"\bEVACUATE\s+NOW\b",
r"\bEVACUATION\s+ORDER\b",
r"\bSET\s+TO\s+GO\b", # "Set to GO" — the standalone-GO idiom, spelled out
]
_SET_PHRASES = [
r"\bLEVEL\s+2\b",
r"\bLEVEL\s+II\b",
r"\bPREPARE\s+TO\s+EVACUATE\b",
r"\bEVACUATION\s+WARNING\b",
r"\bBE\s+READY\s+TO\s+LEAVE\b",
]
_READY_PHRASES = [
r"\bLEVEL\s+1\b",
r"\bLEVEL\s+I\b",
r"\bEVACUATION\s+ADVISORY\b",
]
_PRECEDENCE = ("GO", "SET", "READY")
_PHRASE_RE = {
"GO": re.compile("|".join(_GO_PHRASES), re.IGNORECASE),
"SET": re.compile("|".join(_SET_PHRASES), re.IGNORECASE),
"READY": re.compile("|".join(_READY_PHRASES), re.IGNORECASE),
}
# A bare level-word (GO/SET/READY) counts ONLY when it is:
# 1. standalone (word-boundaried) AND written in the exact uppercase form
# (narrative prose never shouts a whole word in caps: "go to the
# fairgrounds", "set to arrive", "via Go Creek Road" all fail this), AND
# 2. accompanied elsewhere in the same text by other alert/evacuation
# vocabulary, so a bare "GO"/"SET"/"READY" floating in unrelated text
# can't fire on its own.
# This is what lets "LEVEL I SET Alert" resolve as SET (word beats numeral —
# err upward) even though "LEVEL I" alone would read as READY.
_STANDALONE_TOKEN_RE = {
"GO": re.compile(r"\bGO\b"),
"SET": re.compile(r"\bSET\b"),
"READY": re.compile(r"\bREADY\b"),
}
_CONTEXT_CUE_RE = re.compile(
r"evacuat|level|alert|status|notice|prepar|leave|order|warning|advisory",
re.IGNORECASE,
)
def detect_phase(*texts: "str | None") -> "str | None":
"""Scan the given texts for Idaho READY/SET/GO evacuation-phase language.
All provided texts are combined and scanned together (case-insensitive
for the strong phrases). The HIGHEST phase found wins GO > SET > READY
never the first match. Returns None when nothing matches; never
guesses.
"""
combined = "\n".join(t for t in texts if t)
if not combined:
return None
has_cue = bool(_CONTEXT_CUE_RE.search(combined))
for phase in _PRECEDENCE:
if _PHRASE_RE[phase].search(combined):
return phase
if has_cue and _STANDALONE_TOKEN_RE[phase].search(combined):
return phase
return None

View file

@ -96,7 +96,7 @@ def resolve_anchor(
from meshai.persistence import get_db
rows = get_db().execute(
"SELECT name, lat, lon FROM town_anchors "
"WHERE lat IS NOT NULL AND lon IS NOT NULL"
"WHERE lat IS NOT NULL AND lon IS NOT NULL AND enabled = 1"
).fetchall()
best = None
best_d = float("inf")

View file

@ -6,11 +6,19 @@ Message, AMBER, 911 outage, law enforcement, HazMat) from the canonical CAP
Civil alerts carry their signal in the CAP ``headline`` (a plain human
sentence), so unlike the weather formatter, which parses structured
HAZARD.../IMPACT... blocks this formatter is headline-forward:
HAZARD.../IMPACT... blocks this formatter is headline-forward. Real CAP
data has no machine-readable Idaho READY/SET/GO evacuation phase (there is
no ``<responseType>`` in the wild); ``evac_phase.detect_phase`` scans the
headline/CMAMtext/description free text for the phrases agencies actually
use ("Level 3 GO NOW", "Evacuation Warning", ...) so the wire message can
lead with the phase instead of raw CAP jargon:
Line 1: {emoji} {prefix}{event} e.g. "🚨 Evacuation Immediate"
Line 1 (phase detected): {emoji} {prefix}{PHASE} {action}
e.g. "🚨 GO — Leave now"
Line 1 (no phase found unchanged fallback):
{emoji} {prefix}{event} e.g. "🚨 Evacuation Immediate"
Line 2: {area}[ · Until {t} {tz}] areaDesc (first area) + expiry
Line 3: {headline} the operator's message
Line 3: {CMAMtext or headline} the agency's own public alert text
Reuses ``event.data`` (canonical) + ``_budget.fit_to_budget``; does NOT touch
the NWS formatter. ``now`` is a structural seam (not used expiry is absolute).
@ -21,6 +29,7 @@ import zoneinfo
from datetime import datetime
from typing import TYPE_CHECKING
from meshai.notifications.evac_phase import detect_phase
from meshai.notifications.formatters._budget import fit_to_budget
if TYPE_CHECKING:
@ -37,6 +46,13 @@ _CATEGORY_EMOJI = {
"emergency_civil": "⚠️",
}
# Idaho READY/SET/GO — the one-line action tied to each detected phase.
_PHASE_ACTION = {
"READY": "Get prepared",
"SET": "Be ready to leave",
"GO": "Leave now",
}
def format(event: "Event", *, now: float, budget: int) -> str:
"""Render the IPAWS civil-alert wire string from canonical event.data.
@ -54,6 +70,11 @@ def format(event: "Event", *, now: float, budget: int) -> str:
event_type = d.get("event") or "Emergency Alert"
area_desc = d.get("area_desc") or ""
headline = (d.get("headline") or "").strip()
description = d.get("description") or ""
parameters = d.get("parameters") or {}
cmam_text = parameters.get("CMAMtext") or ""
if isinstance(cmam_text, list): # defensive: a repeated valueName collapses to a list
cmam_text = cmam_text[0] if cmam_text else ""
expires_epoch = d.get("expires_at")
prefix = d.get("_ipaws_prefix") or ""
category = d.get("category") or event.category or "emergency_civil"
@ -61,8 +82,20 @@ def format(event: "Event", *, now: float, budget: int) -> str:
emoji = _CATEGORY_EMOJI.get(category, "⚠️")
prefix_seg = f"{prefix}: " if prefix else ""
# Line 1: emoji + prefix + event type
line1 = f"{emoji} {prefix_seg}{event_type}"
# Line 1: lead with the Idaho READY/SET/GO phase when the free text
# actually names one; otherwise keep the raw CAP event string (safe
# fallback — never invent a phase that isn't there).
phase = detect_phase(headline, cmam_text, description)
if phase:
# A detected GO always gets the alarm emoji, regardless of CAP
# category — a confirmed "leave now" evacuation must never render
# with the softer category-based ⚠️ (e.g. emergency_civil).
if phase == "GO":
emoji = "🚨"
action = _PHASE_ACTION[phase]
line1 = f"{emoji} {prefix_seg}{phase}{action}"
else:
line1 = f"{emoji} {prefix_seg}{event_type}"
# Line 2: first area + optional expiry ("Until 4:54 PM MDT")
area = (area_desc or "").split(";")[0].strip()
@ -79,8 +112,9 @@ def format(event: "Event", *, now: float, budget: int) -> str:
else:
line2 = area or time_seg
# Line 3: the headline (the actual civil message)
line3 = headline
# Line 3: the agency's own public alert text (CMAMtext — purpose-written
# for WEA, ~90 chars) when present; else fall back to the headline as before.
line3 = cmam_text.strip() or headline
msg = "\n".join(ln for ln in (line1, line2, line3) if ln)
return fit_to_budget(msg, budget)

View file

@ -151,10 +151,38 @@ def decide(data: dict, *, source: str, now: float) -> GateResult:
contained_pct = data.get("contained_pct")
declared_at_epoch = data.get("declared_at_epoch")
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()
try:
row = conn.execute(
"SELECT current_acres, current_contained_pct, last_broadcast_at, "
"last_broadcast_acres, last_broadcast_contained, tombstoned_at "
"FROM fires WHERE irwin_id = ?", (irwin_id,)).fetchone()
except Exception:
# Defensive: an un-migrated schema without the tombstoned_at column
# (pre-v12) must not crash the decider. Fall back to the narrower
# column set; the tombstone gate below then reads as "not tombstoned"
# for this row rather than raising.
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()
# ── Tombstone mute — durable operator kill-switch ───────────────────────
# A non-null fires.tombstoned_at (the SAME column the dashboard's active
# views already filter on -- /api/env/active WHERE tombstoned_at IS NULL)
# means this incident is closed out / muted. It must never broadcast
# again through ANY path below (new, update, or cooldown-suppress is a
# no-op here anyway), so this check runs BEFORE the row-missing / growth
# / cooldown branches. Defensive: a row without the column, or missing
# the key for any other reason, reads as "not tombstoned" rather than
# raising -- it must never crash the decider.
if row is not None:
try:
tombstoned_at = row["tombstoned_at"]
except (IndexError, KeyError, TypeError):
tombstoned_at = None
if tombstoned_at is not None:
return GateResult(broadcast=False, lifecycle="suppress",
reason=f"incident tombstoned irwin={irwin_id}")
# Stamps every active broadcast (New + Update) carries. category override
# is added only for New (Update keeps the envelope-derived wildfire_incident).

View file

@ -359,6 +359,18 @@ class Dispatcher:
# v0.6-3b: fire toggle uses wfigs adapter_config freshness (0 = disabled)
if fam == "fire":
freshness_s = int(adapter_config.wfigs.freshness_seconds)
elif event.category == "earthquake_event":
# The upstream USGS feed is a rolling PAST-DAY feed
# (2.5_day.geojson), so a quake's age-at-ingest routinely exceeds
# the generic 600s toggle window even for a genuinely first-seen
# event -- every native quake was being silently dropped here
# (measured age-at-ingest 548s-72169s across 12 sampled quakes,
# confirmed against quake_events: 12/12 first-sighted rows with
# last_broadcast_at=NULL). Scoped to the earthquake_event
# CATEGORY, not the "seismic" family/toggle, because that family
# also carries stream_flood_warning/stream_high_water (hydro),
# which must keep using the generic per-toggle freshness_seconds.
freshness_s = int(adapter_config.usgs_quake.freshness_seconds)
else:
freshness_s = int(getattr(tog, "freshness_seconds", 600) or 600)
if event.timestamp and freshness_s > 0:

View file

@ -85,35 +85,192 @@ _GAUGE_SITES_SEED: dict[str, dict[str, Any]] = {
# Idaho + neighbor towns, originally from a hardcoded _TOWN_COORDS table in
# the (since-deleted) Central-envelope adapter-normalizer module.
_TOWN_ANCHORS_SEED: dict[str, dict[str, Any]] = {
"boise": {"lat": 43.6150, "lon": -116.2023, "state": "ID"},
"meridian": {"lat": 43.6121, "lon": -116.3915, "state": "ID"},
"nampa": {"lat": 43.5407, "lon": -116.5635, "state": "ID"},
"caldwell": {"lat": 43.6629, "lon": -116.6874, "state": "ID"},
"idaho falls": {"lat": 43.4666, "lon": -112.0340, "state": "ID"},
"pocatello": {"lat": 42.8713, "lon": -112.4455, "state": "ID"},
"twin falls": {"lat": 42.5630, "lon": -114.4609, "state": "ID"},
"coeur d'alene": {"lat": 47.6777, "lon": -116.7805, "state": "ID"},
"lewiston": {"lat": 46.4165, "lon": -117.0177, "state": "ID"},
"moscow": {"lat": 46.7324, "lon": -117.0002, "state": "ID"},
"sandpoint": {"lat": 48.2766, "lon": -116.5535, "state": "ID"},
"post falls": {"lat": 47.7180, "lon": -116.9516, "state": "ID"},
"hayden": {"lat": 47.7660, "lon": -116.7866, "state": "ID"},
"rathdrum": {"lat": 47.8121, "lon": -116.8950, "state": "ID"},
"plummer": {"lat": 47.3344, "lon": -116.8856, "state": "ID"},
"kellogg": {"lat": 47.5380, "lon": -116.1352, "state": "ID"},
"bonners ferry": {"lat": 48.6914, "lon": -116.3181, "state": "ID"},
"rexburg": {"lat": 43.8260, "lon": -111.7897, "state": "ID"},
"blackfoot": {"lat": 43.1905, "lon": -112.3447, "state": "ID"},
"burley": {"lat": 42.5360, "lon": -113.7928, "state": "ID"},
"jerome": {"lat": 42.7252, "lon": -114.5187, "state": "ID"},
"mountain home": {"lat": 43.1330, "lon": -115.6912, "state": "ID"},
"stanley": {"lat": 44.2160, "lon": -114.9311, "state": "ID"},
"salmon": {"lat": 45.1758, "lon": -113.8957, "state": "ID"},
"mccall": {"lat": 44.9111, "lon": -116.0987, "state": "ID"},
"weiser": {"lat": 44.2510, "lon": -116.9690, "state": "ID"},
"soda springs": {"lat": 42.6543, "lon": -111.6047, "state": "ID"},
"preston": {"lat": 42.0963, "lon": -111.8766, "state": "ID"},
"montpelier": {"lat": 42.3232, "lon": -111.2980, "state": "ID"},
"aberdeen": {"lat": 42.944098, "lon": -112.838381, "state": "ID"},
"albion": {"lat": 42.409808, "lon": -113.580438, "state": "ID"},
"american falls": {"lat": 42.782846, "lon": -112.854211, "state": "ID"},
"ammon": {"lat": 43.474999, "lon": -111.959631, "state": "ID"},
"arbon valley": {"lat": 42.88763, "lon": -112.589386, "state": "ID"},
"arco": {"lat": 43.631893, "lon": -113.301033, "state": "ID"},
"arimo": {"lat": 42.560385, "lon": -112.172927, "state": "ID"},
"ashton": {"lat": 44.073332, "lon": -111.448311, "state": "ID"},
"athol": {"lat": 47.947065, "lon": -116.707958, "state": "ID"},
"avimor": {"lat": 43.776181, "lon": -116.257108, "state": "ID"},
"bancroft": {"lat": 42.720241, "lon": -111.88301, "state": "ID"},
"basalt": {"lat": 43.314443, "lon": -112.165044, "state": "ID"},
"bellevue": {"lat": 43.467894, "lon": -114.254955, "state": "ID"},
"bennington": {"lat": 42.382259, "lon": -111.32098, "state": "ID"},
"blackfoot": {"lat": 43.1905, "lon": -112.3447, "state": "ID"},
"blanchard": {"lat": 48.0137, "lon": -116.996503, "state": "ID"},
"bliss": {"lat": 42.924284, "lon": -114.947516, "state": "ID"},
"boise": {"lat": 43.615, "lon": -116.2023, "state": "ID"},
"bonners ferry": {"lat": 48.6914, "lon": -116.3181, "state": "ID"},
"buhl": {"lat": 42.598362, "lon": -114.759536, "state": "ID"},
"burley": {"lat": 42.536, "lon": -113.7928, "state": "ID"},
"caldwell": {"lat": 43.6629, "lon": -116.6874, "state": "ID"},
"cambridge": {"lat": 44.571748, "lon": -116.678101, "state": "ID"},
"carey": {"lat": 43.312011, "lon": -113.941008, "state": "ID"},
"cascade": {"lat": 44.508761, "lon": -116.043627, "state": "ID"},
"castleford": {"lat": 42.520569, "lon": -114.871806, "state": "ID"},
"challis": {"lat": 44.505779, "lon": -114.228184, "state": "ID"},
"chubbuck": {"lat": 42.926182, "lon": -112.462537, "state": "ID"},
"clark fork": {"lat": 48.148019, "lon": -116.172988, "state": "ID"},
"clifton": {"lat": 42.187286, "lon": -112.004596, "state": "ID"},
"coeur d'alene": {"lat": 47.6777, "lon": -116.7805, "state": "ID"},
"cottonwood": {"lat": 46.051045, "lon": -116.349751, "state": "ID"},
"council": {"lat": 44.733195, "lon": -116.436837, "state": "ID"},
"craigmont": {"lat": 46.24174, "lon": -116.471344, "state": "ID"},
"culdesac": {"lat": 46.374896, "lon": -116.670097, "state": "ID"},
"dalton gardens": {"lat": 47.733412, "lon": -116.767873, "state": "ID"},
"dayton": {"lat": 42.111246, "lon": -111.984615, "state": "ID"},
"deary": {"lat": 46.800586, "lon": -116.557368, "state": "ID"},
"declo": {"lat": 42.519599, "lon": -113.628732, "state": "ID"},
"dietrich": {"lat": 42.912796, "lon": -114.266289, "state": "ID"},
"donnelly": {"lat": 44.733391, "lon": -116.086821, "state": "ID"},
"dover": {"lat": 48.259156, "lon": -116.609843, "state": "ID"},
"downey": {"lat": 42.428828, "lon": -112.123303, "state": "ID"},
"driggs": {"lat": 43.729865, "lon": -111.104319, "state": "ID"},
"dubois": {"lat": 44.171161, "lon": -112.228278, "state": "ID"},
"eagle": {"lat": 43.693423, "lon": -116.345989, "state": "ID"},
"east hope": {"lat": 48.240895, "lon": -116.28941, "state": "ID"},
"eden": {"lat": 42.605309, "lon": -114.209087, "state": "ID"},
"emmett": {"lat": 43.869228, "lon": -116.491336, "state": "ID"},
"fairfield": {"lat": 43.348045, "lon": -114.800826, "state": "ID"},
"fernwood": {"lat": 47.115099, "lon": -116.386484, "state": "ID"},
"filer": {"lat": 42.56789, "lon": -114.611471, "state": "ID"},
"firth": {"lat": 43.305788, "lon": -112.183454, "state": "ID"},
"fort hall": {"lat": 43.014547, "lon": -112.45787, "state": "ID"},
"franklin": {"lat": 42.009503, "lon": -111.802183, "state": "ID"},
"fruitland": {"lat": 44.020412, "lon": -116.922109, "state": "ID"},
"garden city": {"lat": 43.668297, "lon": -116.294389, "state": "ID"},
"garden valley": {"lat": 44.083404, "lon": -115.958464, "state": "ID"},
"genesee": {"lat": 46.551608, "lon": -116.928389, "state": "ID"},
"georgetown": {"lat": 42.479083, "lon": -111.363497, "state": "ID"},
"glenns ferry": {"lat": 42.949908, "lon": -115.308185, "state": "ID"},
"gooding": {"lat": 42.937048, "lon": -114.713188, "state": "ID"},
"grace": {"lat": 42.575182, "lon": -111.729771, "state": "ID"},
"grand view": {"lat": 42.985123, "lon": -116.09368, "state": "ID"},
"grangeville": {"lat": 45.925826, "lon": -116.121916, "state": "ID"},
"greenleaf": {"lat": 43.672607, "lon": -116.821443, "state": "ID"},
"groveland": {"lat": 43.223483, "lon": -112.37547, "state": "ID"},
"hagerman": {"lat": 42.816016, "lon": -114.897681, "state": "ID"},
"hailey": {"lat": 43.512674, "lon": -114.299499, "state": "ID"},
"hammett": {"lat": 42.944114, "lon": -115.465521, "state": "ID"},
"hansen": {"lat": 42.531365, "lon": -114.301177, "state": "ID"},
"harrison": {"lat": 47.469582, "lon": -116.808336, "state": "ID"},
"hauser": {"lat": 47.773694, "lon": -117.008, "state": "ID"},
"hayden": {"lat": 47.766, "lon": -116.7866, "state": "ID"},
"hayden lake": {"lat": 47.76446, "lon": -116.756097, "state": "ID"},
"hazelton": {"lat": 42.59548, "lon": -114.136614, "state": "ID"},
"heyburn": {"lat": 42.559982, "lon": -113.762067, "state": "ID"},
"hidden springs": {"lat": 43.716541, "lon": -116.259415, "state": "ID"},
"hollister": {"lat": 42.352911, "lon": -114.583846, "state": "ID"},
"homedale": {"lat": 43.615937, "lon": -116.939029, "state": "ID"},
"horseshoe bend": {"lat": 43.916085, "lon": -116.199236, "state": "ID"},
"idaho city": {"lat": 43.827844, "lon": -115.830474, "state": "ID"},
"idaho falls": {"lat": 43.4666, "lon": -112.034, "state": "ID"},
"inkom": {"lat": 42.796548, "lon": -112.254625, "state": "ID"},
"iona": {"lat": 43.527007, "lon": -111.930914, "state": "ID"},
"irwin": {"lat": 43.403359, "lon": -111.279513, "state": "ID"},
"jerome": {"lat": 42.7252, "lon": -114.5187, "state": "ID"},
"juliaetta": {"lat": 46.574737, "lon": -116.70808, "state": "ID"},
"kamiah": {"lat": 46.226796, "lon": -116.028303, "state": "ID"},
"kellogg": {"lat": 47.538, "lon": -116.1352, "state": "ID"},
"kendrick": {"lat": 46.614183, "lon": -116.661272, "state": "ID"},
"ketchum": {"lat": 43.687718, "lon": -114.380069, "state": "ID"},
"kimberly": {"lat": 42.534299, "lon": -114.369931, "state": "ID"},
"kooskia": {"lat": 46.141687, "lon": -115.973646, "state": "ID"},
"kootenai": {"lat": 48.311831, "lon": -116.517128, "state": "ID"},
"kuna": {"lat": 43.469607, "lon": -116.424153, "state": "ID"},
"laclede": {"lat": 48.167129, "lon": -116.751565, "state": "ID"},
"lapwai": {"lat": 46.403715, "lon": -116.804223, "state": "ID"},
"lava hot springs": {"lat": 42.620107, "lon": -112.009902, "state": "ID"},
"lewiston": {"lat": 46.4165, "lon": -117.0177, "state": "ID"},
"lewisville": {"lat": 43.695208, "lon": -112.013232, "state": "ID"},
"lincoln": {"lat": 43.51825, "lon": -111.969215, "state": "ID"},
"mackay": {"lat": 43.911996, "lon": -113.612728, "state": "ID"},
"malad city": {"lat": 42.189909, "lon": -112.249688, "state": "ID"},
"marsing": {"lat": 43.54636, "lon": -116.810422, "state": "ID"},
"mccall": {"lat": 44.9111, "lon": -116.0987, "state": "ID"},
"mccammon": {"lat": 42.648236, "lon": -112.189394, "state": "ID"},
"melba": {"lat": 43.373633, "lon": -116.531933, "state": "ID"},
"menan": {"lat": 43.721791, "lon": -111.992353, "state": "ID"},
"meridian": {"lat": 43.6121, "lon": -116.3915, "state": "ID"},
"middleton": {"lat": 43.711593, "lon": -116.615008, "state": "ID"},
"montpelier": {"lat": 42.3232, "lon": -111.298, "state": "ID"},
"moreland": {"lat": 43.21948, "lon": -112.437772, "state": "ID"},
"moscow": {"lat": 46.7324, "lon": -117.0002, "state": "ID"},
"mountain home": {"lat": 43.133, "lon": -115.6912, "state": "ID"},
"mountain home afb": {"lat": 43.049186, "lon": -115.86586, "state": "ID"},
"moyie springs": {"lat": 48.724746, "lon": -116.195421, "state": "ID"},
"mud lake": {"lat": 43.842855, "lon": -112.479504, "state": "ID"},
"mullan": {"lat": 47.468759, "lon": -115.796351, "state": "ID"},
"nampa": {"lat": 43.5407, "lon": -116.5635, "state": "ID"},
"new meadows": {"lat": 44.971335, "lon": -116.285195, "state": "ID"},
"new plymouth": {"lat": 43.970417, "lon": -116.818781, "state": "ID"},
"newdale": {"lat": 43.886385, "lon": -111.603888, "state": "ID"},
"nezperce": {"lat": 46.233582, "lon": -116.241418, "state": "ID"},
"notus": {"lat": 43.726863, "lon": -116.800432, "state": "ID"},
"oakley": {"lat": 42.24206, "lon": -113.883058, "state": "ID"},
"oldtown": {"lat": 48.182488, "lon": -117.018005, "state": "ID"},
"orofino": {"lat": 46.484943, "lon": -116.253027, "state": "ID"},
"osburn": {"lat": 47.505731, "lon": -116.000709, "state": "ID"},
"paris": {"lat": 42.227978, "lon": -111.402424, "state": "ID"},
"parker": {"lat": 43.958405, "lon": -111.759206, "state": "ID"},
"parma": {"lat": 43.786284, "lon": -116.942491, "state": "ID"},
"paul": {"lat": 42.605496, "lon": -113.784487, "state": "ID"},
"payette": {"lat": 44.080093, "lon": -116.926852, "state": "ID"},
"pierce": {"lat": 46.495335, "lon": -115.803292, "state": "ID"},
"pinehurst": {"lat": 47.536314, "lon": -116.231746, "state": "ID"},
"plummer": {"lat": 47.3344, "lon": -116.8856, "state": "ID"},
"pocatello": {"lat": 42.8713, "lon": -112.4455, "state": "ID"},
"ponderay": {"lat": 48.30478, "lon": -116.536645, "state": "ID"},
"post falls": {"lat": 47.718, "lon": -116.9516, "state": "ID"},
"potlatch": {"lat": 46.923493, "lon": -116.897713, "state": "ID"},
"preston": {"lat": 42.0963, "lon": -111.8766, "state": "ID"},
"priest river": {"lat": 48.18336, "lon": -116.884354, "state": "ID"},
"rathdrum": {"lat": 47.8121, "lon": -116.895, "state": "ID"},
"rexburg": {"lat": 43.826, "lon": -111.7897, "state": "ID"},
"richfield": {"lat": 43.05164, "lon": -114.155942, "state": "ID"},
"rigby": {"lat": 43.673587, "lon": -111.913525, "state": "ID"},
"riggins": {"lat": 45.420591, "lon": -116.317636, "state": "ID"},
"ririe": {"lat": 43.632494, "lon": -111.771717, "state": "ID"},
"riverside": {"lat": 43.196554, "lon": -112.435625, "state": "ID"},
"roberts": {"lat": 43.720488, "lon": -112.128868, "state": "ID"},
"robie creek": {"lat": 43.667649, "lon": -116.015203, "state": "ID"},
"rockford": {"lat": 43.189235, "lon": -112.530618, "state": "ID"},
"rockford bay": {"lat": 47.508637, "lon": -116.886536, "state": "ID"},
"rockland": {"lat": 42.573157, "lon": -112.87453, "state": "ID"},
"rupert": {"lat": 42.618936, "lon": -113.673967, "state": "ID"},
"salmon": {"lat": 45.1758, "lon": -113.8957, "state": "ID"},
"sandpoint": {"lat": 48.2766, "lon": -116.5535, "state": "ID"},
"shelley": {"lat": 43.379538, "lon": -112.126098, "state": "ID"},
"shoshone": {"lat": 42.936185, "lon": -114.404747, "state": "ID"},
"silverton": {"lat": 47.495681, "lon": -115.960513, "state": "ID"},
"smelterville": {"lat": 47.542423, "lon": -116.177448, "state": "ID"},
"soda springs": {"lat": 42.6543, "lon": -111.6047, "state": "ID"},
"spirit lake": {"lat": 47.965799, "lon": -116.869831, "state": "ID"},
"st. anthony": {"lat": 43.964839, "lon": -111.685049, "state": "ID"},
"st. maries": {"lat": 47.314589, "lon": -116.572235, "state": "ID"},
"stanley": {"lat": 44.216, "lon": -114.9311, "state": "ID"},
"star": {"lat": 43.702788, "lon": -116.491025, "state": "ID"},
"sugar city": {"lat": 43.87582, "lon": -111.751032, "state": "ID"},
"sun valley": {"lat": 43.683852, "lon": -114.334203, "state": "ID"},
"swan valley": {"lat": 43.442575, "lon": -111.324544, "state": "ID"},
"teton": {"lat": 43.887773, "lon": -111.672254, "state": "ID"},
"tetonia": {"lat": 43.814578, "lon": -111.158664, "state": "ID"},
"troy": {"lat": 46.737981, "lon": -116.773154, "state": "ID"},
"twin falls": {"lat": 42.563, "lon": -114.4609, "state": "ID"},
"tyhee": {"lat": 42.954001, "lon": -112.456199, "state": "ID"},
"ucon": {"lat": 43.593538, "lon": -111.959358, "state": "ID"},
"victor": {"lat": 43.60155, "lon": -111.110822, "state": "ID"},
"wallace": {"lat": 47.473578, "lon": -115.922542, "state": "ID"},
"weippe": {"lat": 46.37825, "lon": -115.938844, "state": "ID"},
"weiser": {"lat": 44.251, "lon": -116.969, "state": "ID"},
"wendell": {"lat": 42.77468, "lon": -114.70296, "state": "ID"},
"weston": {"lat": 42.036209, "lon": -111.977799, "state": "ID"},
"wilder": {"lat": 43.678388, "lon": -116.907585, "state": "ID"},
"winchester": {"lat": 46.240812, "lon": -116.624137, "state": "ID"},
"worley": {"lat": 47.400533, "lon": -116.919254, "state": "ID"},
}

View file

@ -36,6 +36,36 @@ _TELEMETRY_MIN_INTERVAL_SECONDS = 300
# dropped from the auto-poll rotation (a manual "Poll now" un-sticks it).
_TELEMETRY_MAX_FAILURES = 3
# --- Companion-link keepalive tuning --------------------------------------
# MeshMonitor's MeshCore vnode (the shared companion-link server meshai
# attaches to) reaps any client idle >5 min, where "idle" means no bytes seen
# FROM the client — a periodic LOCAL query resets that clock. 120s is well
# inside the 300s reaper window with margin to spare.
_KEEPALIVE_INTERVAL_SECONDS = 120
# --- Reconnect persistence ("0 = unlimited" sentinel) ----------------------
# config.py documents meshcore_max_reconnect_attempts as "0 = unlimited", but
# that sentinel was never implemented here — the value was passed straight
# through to the meshcore lib's ConnectionManager, whose retry loop is
# `while self._reconnect_attempts < self.max_reconnect_attempts`. Taken
# literally, 0 means ZERO attempts (immediate give-up), the opposite of
# "unlimited", and any small bounded value (the shipped default is 5, at the
# lib's flat 1s-per-attempt cadence) exhausts after ~5 seconds and then the
# link stays down PERMANENTLY — there is no external supervisor for MeshCore
# (see main.py's watchdog guard: "MeshCoreTransport manages its own
# reconnect via the meshcore lib's auto_reconnect parameter"), so nothing
# ever notices and retries again after that. A radio/vnode bounce longer
# than ~5s (e.g. the 2026-08-02 device-perm heal test) killed MeshCore for
# good until a manual container restart.
#
# Fix: honor the documented sentinel for real. connect() below translates a
# configured 0 into this effectively-unbounded count, so the lib's own
# proven-safe retry loop (still local TCP only, still ~1 attempt/sec, still
# WITHOUT re-sending the connect-time self-advert — see
# _post_reconnect_setup_async) just keeps going until the vnode/radio comes
# back, no matter how long the outage lasts.
_MC_RECONNECT_ATTEMPTS_UNLIMITED = 2_147_483_647
# Numeric Cayenne-LPP type id → decoded field name. Ids not in this map are
# passed through as ``lpp_<id>`` so nothing is silently dropped.
_LPP_ID_TO_FIELD = {
@ -132,6 +162,8 @@ class MeshCoreTransport(MeshTransport):
self._advert_task = None
# asyncio.Task handle for the telemetry auto-poll loop; None when inactive.
self._telemetry_task = None
# asyncio.Task handle for the companion-link keepalive loop; None when inactive.
self._keepalive_task = None
# Telemetry availability/bookkeeping (shared by poller + on-demand):
# _telemetry_cache: contact-id -> {contact, data, polled_at, available}
# _telemetry_failures: contact-id -> consecutive-timeout count
@ -577,15 +609,37 @@ class MeshCoreTransport(MeshTransport):
)
return acked or (not result.is_error())
def _resolve_mc_channel_idx(self, meshcore_channel: str) -> Optional[int]:
"""Resolve a config channel name to the companion's channel slot.
Tries an exact match first (fast path, preserves existing behavior
for e.g. ``#bot``). Falls back to a match that ignores a single
leading ``#`` and case, since after the radio moved to MeshMonitor's
vnode the companion enumerates region channels WITHOUT the leading
``#`` that meshai's region_routes config still carries (e.g. config
``#sc-id-aida`` vs. companion ``sc-id-aida``) — same channel/key,
just a display-name difference upstream.
"""
idx = self._chan_name_to_idx.get(meshcore_channel)
if idx is not None:
return idx
canon = meshcore_channel[1:] if meshcore_channel.startswith("#") else meshcore_channel
canon = canon.casefold()
for name, slot in self._chan_name_to_idx.items():
name_canon = name[1:] if name.startswith("#") else name
if name_canon.casefold() == canon:
return slot
return None
async def _do_mc_broadcast_async(self, text: str, meshcore_channel: str) -> bool:
"""Channel broadcast on the MC loop (replaces send_message() broadcast branch)."""
if self._mc is None:
return False
idx = self._chan_name_to_idx.get(meshcore_channel)
idx = self._resolve_mc_channel_idx(meshcore_channel)
if idx is None:
# Lazy async re-enumeration (no _run_coro deadlock risk).
await self._enumerate_channels_async()
idx = self._chan_name_to_idx.get(meshcore_channel)
idx = self._resolve_mc_channel_idx(meshcore_channel)
if idx is None:
logger.warning("MC channel '%s' not on companion; skipping", meshcore_channel)
return False
@ -1776,6 +1830,66 @@ class MeshCoreTransport(MeshTransport):
if task is not None and self._loop is not None and self._loop.is_running():
self._loop.call_soon_threadsafe(task.cancel)
# ------------------------------------------------------------------
# Companion-link keepalive (Task on the dedicated loop)
# ------------------------------------------------------------------
async def _keepalive_loop(self) -> None:
"""Quiet LOCAL companion-link keepalive (Task on the dedicated loop).
MeshMonitor's MeshCore vnode disconnects any client idle >5 min,
where "idle" means no bytes seen FROM the client its
``lastActivity`` only updates on data we send it, never on data it
sends us. The self-advert (every 24h by default) and telemetry poll
(30 min default, and only when contacts are configured) are both far
too infrequent to keep that clock fresh, so the link was silently
reaped and never recovered (``meshcore_auto_reconnect`` is the
recovery safety net; this loop is the prevention).
Every ``_KEEPALIVE_INTERVAL_SECONDS`` (while connected), issues
``commands.get_time()`` a single-byte companion opcode (CMD 0x05)
that reads the node's own onboard clock and returns CURRENT_TIME.
It carries no destination/contact and has no mesh-routing semantics
(unlike send_advert/send_msg/send_chan_msg), so the firmware answers
it purely locally over the companion link it does not key the
radio or emit an RF packet. Runs directly on the MC loop (NOT
through the send queue/pacing it is a device-info query, not a
mesh send, so it should never wait behind or delay a real send).
Stops on CancelledError (disconnect). A transient query failure is
logged and ignored the loop keeps ticking every interval either
way, since the point is resetting the vnode's clock on our next
successful frame, not the query result itself.
"""
try:
while True:
await asyncio.sleep(_KEEPALIVE_INTERVAL_SECONDS)
if not self._connected or self._mc is None:
return
try:
await self._mc.commands.get_time()
logger.debug("MC: companion-link keepalive query sent")
except Exception as exc:
logger.debug("MC: keepalive get_time failed (non-fatal): %s", exc)
except asyncio.CancelledError:
logger.debug("MC: keepalive task cancelled")
raise
def _schedule_keepalive(self) -> None:
"""Create the keepalive asyncio.Task on the dedicated loop (thread-safe)."""
def _arm() -> None:
self._keepalive_task = asyncio.get_event_loop().create_task(
self._keepalive_loop()
)
self._loop.call_soon_threadsafe(_arm)
def _cancel_keepalive(self) -> None:
"""Cancel the keepalive task (thread-safe). Called at disconnect."""
task = self._keepalive_task
self._keepalive_task = None
if task is not None and self._loop is not None and self._loop.is_running():
self._loop.call_soon_threadsafe(task.cancel)
# ------------------------------------------------------------------
# Internal coroutines (run on the dedicated loop)
# ------------------------------------------------------------------
@ -1885,6 +1999,16 @@ class MeshCoreTransport(MeshTransport):
ble_address = getattr(self.config, "meshcore_ble_address", "")
auto_reconnect = getattr(self.config, "meshcore_auto_reconnect", True)
max_attempts = getattr(self.config, "meshcore_max_reconnect_attempts", 5)
if max_attempts <= 0:
# Documented sentinel (config.py: "0 = unlimited") — see
# _MC_RECONNECT_ATTEMPTS_UNLIMITED's docstring for why this was
# never actually unlimited before and why translating it here is
# the fix.
logger.info(
"MeshCoreTransport: meshcore_max_reconnect_attempts=%s (unlimited) -> %d",
max_attempts, _MC_RECONNECT_ATTEMPTS_UNLIMITED,
)
max_attempts = _MC_RECONNECT_ATTEMPTS_UNLIMITED
# Human-readable target for logging — from the same descriptor that
# self_info() reports, so the log and the API never disagree.
@ -1963,6 +2087,12 @@ class MeshCoreTransport(MeshTransport):
if telem_interval > 0:
self._schedule_telemetry_poll()
# Arm the quiet local companion-link keepalive — unconditional (not
# a mesh operation, no config gate): protects against the
# MeshMonitor vnode's 5-min idle reaper regardless of advert/
# telemetry cadence.
self._schedule_keepalive()
logger.info(
"MeshCoreTransport: connected as %s (pubkey %s)",
self._self_info.get("name", "unknown"),
@ -1971,9 +2101,10 @@ class MeshCoreTransport(MeshTransport):
def disconnect(self) -> None:
"""Disconnect and stop the event loop thread."""
# Cancel periodic advert + telemetry poll before tearing down the loop.
# Cancel periodic advert + telemetry poll + keepalive before tearing down the loop.
self._cancel_periodic_advert()
self._cancel_telemetry_poll()
self._cancel_keepalive()
if self._mc is not None:
try:
self._run_coro(self._do_disconnect(), timeout=10.0)
@ -2078,12 +2209,12 @@ class MeshCoreTransport(MeshTransport):
)
return True
# Resolve NAME → slot against the live companion channel table.
idx = self._chan_name_to_idx.get(meshcore_channel)
idx = self._resolve_mc_channel_idx(meshcore_channel)
if idx is None:
# One lazy re-enumeration in case the table changed since
# connect (e.g. a channel was provisioned after startup).
self._enumerate_channels()
idx = self._chan_name_to_idx.get(meshcore_channel)
idx = self._resolve_mc_channel_idx(meshcore_channel)
if idx is None:
# Never blind-send to a guessed slot.
logger.warning(
@ -2263,10 +2394,65 @@ class MeshCoreTransport(MeshTransport):
self._connected = False
logger.warning("MeshCoreTransport: DISCONNECTED event received")
async def _post_reconnect_setup_async(self) -> None:
"""Redo connect()'s LOCAL post-connect setup after an auto-reconnect.
connect() does this setup once, on the initial connect: rebuild
``_chan_name_to_idx`` (so channel-name broadcasts can resolve a
slot) and arm the companion-link keepalive (so MeshMonitor's vnode
doesn't reap the link again at 5 min idle). The meshcore lib's
auto-reconnect only re-establishes the socket and fires CONNECTED
(-> ``_on_connect_event``) it does not repeat that setup, so a
reconnected link was left with an empty channel table and no
keepalive until the reaper cut it again.
Both steps here are local companion queries/timers only
``_enumerate_channels_async`` calls ``get_channel()`` and the
keepalive calls ``get_time()`` (see their docstrings); neither
keys the radio or emits an RF packet. This deliberately excludes
connect()'s ``send_advert()`` — that IS a transmission, and must
stay confined to the initial connect() path, never replayed on
reconnect.
Keepalive re-arm is cancel-then-schedule (idempotent) so it never
double-schedules the task.
Runs as a fire-and-forget task on the dedicated MC loop (see
``_on_connect_event``) rather than being awaited inline via
``_run_coro``: the meshcore lib invokes ``_on_connect_event`` from
within that same loop (like ``_on_new_contact``), so a blocking
``_run_coro().result()`` call here would deadlock it.
"""
try:
await self._enumerate_channels_async()
except Exception:
logger.warning(
"MeshCore: post-reconnect channel re-enumeration failed", exc_info=True
)
try:
self._cancel_keepalive()
self._schedule_keepalive()
except Exception:
logger.warning(
"MeshCore: post-reconnect keepalive re-arm failed", exc_info=True
)
def _on_connect_event(self, event=None) -> None:
"""Track link state: CONNECTED (auto-reconnect succeeded)."""
"""Track link state: CONNECTED (auto-reconnect succeeded).
Schedules ``_post_reconnect_setup_async`` fire-and-forget on the
dedicated MC loop see that method's docstring for why this must
not block (``_run_coro`` would deadlock from inside this callback,
exactly as noted in ``_on_new_contact``).
"""
self._connected = True
logger.info("MeshCoreTransport: CONNECTED event received")
try:
loop = getattr(self, "_loop", None)
if loop is not None and loop.is_running():
asyncio.run_coroutine_threadsafe(self._post_reconnect_setup_async(), loop)
except Exception:
logger.debug("MeshCore: scheduling post-reconnect setup failed", exc_info=True)
# ------------------------------------------------------------------
# Node identity / topology (MeshTransport abstract methods)

View file

@ -86,6 +86,7 @@ def test_per_adapter_list(client):
"regional_centroid", "regional_radius_mi",
"broadcast_pager_alerts", "global_mag_floor",
"regional_mag_floor", "escalate_mag_floor",
"freshness_seconds",
}

View file

@ -349,6 +349,12 @@ def test_decider_uses_own_table_not_nws(_isolate_meshai_db):
# ============================================================
def test_formatter_renders_civil_alert():
"""Idaho CEM headline ("Wildfire Immediate Evacuation Alert") names the GO
phase explicitly ("Immediate Evacuation"), so line 1 leads with GO instead
of the raw CAP event string the whole point of this feature. Category
is emergency_civil (normally ), but a detected GO forces the alarm
emoji regardless of category a confirmed "leave now" must never render
with the softer warning icon."""
from meshai.notifications.formatters.ipaws import format as ipaws_format
a = IPAWSAlertsAdapter(_config())
raw = a._parse_cap(_fx("eas_idaho_cem.xml"), "16")
@ -357,20 +363,73 @@ def test_formatter_renders_civil_alert():
wire = ipaws_format(ev, now=1000.0, budget=200)
assert len(wire) <= 200
lines = wire.split("\n")
assert lines[0].startswith("⚠️") # civil -> warning emoji
assert "Civil Emergency Message" in lines[0]
assert lines[0].startswith("🚨") # GO overrides civil's ⚠️ -> alarm emoji
assert "GO" in lines[0] and "Leave now" in lines[0]
assert "Boundary County" in wire
assert "evacuation" in wire.lower() # headline carried the signal
def test_formatter_evacuation_emoji():
"""Oregon EVI has no phase language in its headline, but its CMAMtext
parameter ("Level 3 GO NOW evacuation notice...") does, and that text is
now parsed and preferred for line 3 so GO must win here too."""
from meshai.notifications.formatters.ipaws import format as ipaws_format
a = IPAWSAlertsAdapter(_config())
raw = a._parse_cap(_fx("eas_oregon_evi.xml"), "41")
ev = a.to_event(raw)
wire = ipaws_format(ev, now=1000.0, budget=200)
assert wire.startswith("🚨") # evacuation -> alarm emoji
assert "Evacuation Immediate" in wire
lines = wire.split("\n")
assert "GO" in lines[0] and "Leave now" in lines[0]
assert "Level 3 GO NOW" in wire # CMAMtext carried into line 3
def test_formatter_full_three_line_go_rendering():
"""Full 3-line render for a real GO alert (Oregon EVI): phase-led line 1,
area+expiry line 2, CMAMtext line 3 exact text, not just substrings."""
from meshai.notifications.formatters.ipaws import format as ipaws_format
a = IPAWSAlertsAdapter(_config())
raw = a._parse_cap(_fx("eas_oregon_evi.xml"), "41")
ev = a.to_event(raw)
ev.data["_ipaws_prefix"] = ""
wire = ipaws_format(ev, now=1000.0, budget=200)
lines = wire.split("\n")
assert lines[0] == "🚨 GO — Leave now"
assert lines[1] == "Jackson County · Until 8:50 AM MDT"
assert lines[2] == (
"Wildfire Alert- Level 3 GO NOW evacuation notice is UPGRADED for JAC-126"
)
def test_formatter_no_phase_fallback_unchanged():
"""When no READY/SET/GO language is present anywhere in headline/CMAMtext/
description, line 1 keeps the CURRENT raw-event-string behaviour exactly
the safe fallback this feature must never break."""
from meshai.notifications.formatters.ipaws import format as ipaws_format
from meshai.notifications.events import make_event
canonical = _canonical(
event="Civil Emergency Message",
headline="Boil water advisory issued for the district",
description="A water main break has contaminated the supply.",
parameters={},
expires_at=None,
)
ev = make_event(
source="ipaws",
category="emergency_civil",
severity="priority",
title=canonical["headline"],
summary=canonical["headline"],
body=canonical["description"],
data=canonical,
)
ev.data["_ipaws_prefix"] = ""
wire = ipaws_format(ev, now=1000.0, budget=200)
lines = wire.split("\n")
assert lines[0] == "⚠️ Civil Emergency Message" # unchanged fallback (no phase)
assert lines[1] == "Boundary County"
assert lines[2] == "Boil water advisory issued for the district"
# ============================================================
@ -386,3 +445,36 @@ def test_coverage_state_fips_for_bbox():
resolved = resolve_adapter_coverage("ipaws", idaho_bbox, "native")
assert "16" in resolved["state_fips"]
assert resolved["bbox"] == [round(c, 6) for c in idaho_bbox]
# ============================================================
# stage-2 failure negative cache (no re-hammering FEMA)
# ============================================================
def test_stage2_forbidden_is_negative_cached():
"""A 403 on a stage-2 detail URL is remembered so the next poll tick does
NOT re-fetch it, while a sibling URL that succeeds is fetched every pass."""
from urllib.error import HTTPError
counts: dict[str, int] = {}
def _urlopen(req, timeout=None):
url = req.full_url
counts[url] = counts.get(url, 0) + 1
if url.endswith("/feed"):
return _FakeResp(_fx("feed.xml"))
if url.endswith("/eas/300130859542"): # Idaho CEM -> forbidden
raise HTTPError(url, 403, "Forbidden", {}, None)
if url.endswith("/eas/300130856756"): # Oregon EVI -> ok
return _FakeResp(_fx("eas_oregon_evi.xml"))
raise AssertionError(f"unexpected fetch: {url}")
a = IPAWSAlertsAdapter(_config())
with patch("meshai.env.ipaws.urlopen", _urlopen):
a._fetch()
a._fetch()
cem = next(u for u in counts if u.endswith("/eas/300130859542"))
evi = next(u for u in counts if u.endswith("/eas/300130856756"))
assert counts[cem] == 1 # 403 -> negative-cached, not re-fetched
assert counts[evi] == 2 # success -> fetched on each pass

View file

@ -200,3 +200,331 @@ def test_to_event_missing_properties_returns_event(adapter):
def test_to_event_does_not_raise_on_corrupted_dict(adapter):
"""Corrupted dict returns None without raising."""
assert adapter.to_event({"garbage": True}) is None
# ============================================================
# EventType / EventSubType / Cause -> sub_type MAPPING TESTS
#
# Drives real upstream-shaped ITD 511 v2 payloads through the actual
# _parse_event() -> to_event() pipeline (not the make_511_event() stored-
# dict helper) so the EventSubType/Cause capture at parse time and the
# sub_type precedence in to_event() are both exercised end to end.
# ============================================================
def _itd_raw(
event_id="RW-1001",
event_type="roadwork",
event_sub_type="roadConstruction",
cause="Construction",
description="Paving operations, expect delays",
roadway="I-84",
is_full_closure=False,
lat=43.5,
lon=-116.2,
):
"""Realistic raw ITD 511 v2 API item shape (PascalCase fields)."""
return {
"EventId": event_id,
"EventType": event_type,
"EventSubType": event_sub_type,
"Cause": cause,
"RoadwayName": roadway,
"Description": description,
"Latitude": lat,
"Longitude": lon,
"IsFullClosure": is_full_closure,
"LastUpdated": "2026-08-01T00:00:00Z",
}
def _through_pipeline(adapter, raw_item):
"""Parse a raw ITD item then translate to a pipeline Event, mirroring
the real adapter flow (_fetch -> _parse_event -> get_events -> to_event)."""
stored = adapter._parse_event(raw_item, time.time())
assert stored is not None
return adapter.to_event(stored)
def test_mapping_roadwork_not_full_closure_is_road_works(adapter):
"""Routine roadwork, not a full closure -> sub_type 'road_works' (suppressed)."""
raw = _itd_raw(event_type="roadwork", is_full_closure=False)
event = _through_pipeline(adapter, raw)
assert event is not None
assert event.data["sub_type"] == "road_works"
def test_mapping_roadwork_with_full_closure_is_road_closed(adapter):
"""Construction-caused FULL closure is still genuine news -> 'road_closed',
and must NOT be suppressed like routine roadwork."""
raw = _itd_raw(
event_type="roadwork",
event_sub_type="bridgeConstruction",
description="Bridge replacement in progress",
is_full_closure=True,
)
event = _through_pipeline(adapter, raw)
assert event is not None
assert event.data["sub_type"] == "road_closed"
def test_mapping_accidents_and_incidents_is_incident(adapter):
"""EventType accidentsAndIncidents -> sub_type 'incident'."""
raw = _itd_raw(
event_id="INC-2001",
event_type="accidentsAndIncidents",
event_sub_type="crash",
cause="Collision",
description="Two vehicle collision blocking right lane",
is_full_closure=False,
)
event = _through_pipeline(adapter, raw)
assert event is not None
assert event.data["sub_type"] == "incident"
def test_mapping_closures_is_road_closed(adapter):
"""EventType closures -> sub_type 'road_closed'."""
raw = _itd_raw(
event_id="CL-3001",
event_type="closures",
event_sub_type="bridgeClosure",
cause="Maintenance",
description="Bridge out of service",
is_full_closure=False,
)
event = _through_pipeline(adapter, raw)
assert event is not None
assert event.data["sub_type"] == "road_closed"
def test_mapping_special_events_is_incident(adapter):
"""EventType specialEvents -> sub_type 'incident' (not suppressed)."""
raw = _itd_raw(
event_id="SE-4001",
event_type="specialEvents",
event_sub_type="parade",
cause=None,
description="Downtown parade route",
is_full_closure=False,
)
event = _through_pipeline(adapter, raw)
assert event is not None
assert event.data["sub_type"] == "incident"
def test_mapping_unknown_or_missing_event_type_is_incident(adapter):
"""Missing/unrecognized EventType fails open to 'incident' (not suppressed)."""
raw = {
"EventId": "UNK-5001",
"RoadwayName": "SH-21",
"Description": "Unusual event",
"Latitude": 43.5,
"Longitude": -115.5,
}
event = _through_pipeline(adapter, raw)
assert event is not None
assert event.data["sub_type"] == "incident"
def test_mapping_event_subtype_construction_overrides_non_roadwork_eventtype(adapter):
"""EventSubType containing 'construction' forces 'road_works' even when
EventType itself is not literally 'roadwork'."""
raw = _itd_raw(
event_id="SE-6001",
event_type="specialEvents",
event_sub_type="bridgeConstruction",
cause="Construction",
description="Bridge deck construction nearby",
is_full_closure=False,
)
event = _through_pipeline(adapter, raw)
assert event is not None
assert event.data["sub_type"] == "road_works"
def test_mapping_cause_is_carried_through_not_none(adapter):
"""Cause from the upstream payload reaches canonical_data, not hardcoded None."""
raw = _itd_raw(cause="Weather")
event = _through_pipeline(adapter, raw)
assert event is not None
assert event.data["cause"] == "Weather"
assert event.data["cause"] is not None
def test_mapping_missing_cause_is_none(adapter):
"""No Cause field upstream -> canonical cause is None, not a crash."""
raw = _itd_raw(cause=None)
del raw["Cause"]
event = _through_pipeline(adapter, raw)
assert event is not None
assert event.data["cause"] is None
# ============================================================
# END-TO-END GATING TEST — proves a real roadwork event, run through the
# actual adapter pipeline, is suppressed by gating/incident.py's work-zone
# rule, while a real crash event from the same pipeline still broadcasts.
# ============================================================
def test_roadwork_event_suppressed_by_incident_gate_end_to_end(adapter):
"""A real ITD roadwork payload, parsed and translated by the adapter,
must be suppressed at the gating layer (not just mapped to the right
sub_type in isolation)."""
from meshai.notifications.gating.incident import decide
raw = _itd_raw(event_id="RW-7001", event_type="roadwork", is_full_closure=False)
event = _through_pipeline(adapter, raw)
assert event is not None
assert event.data["sub_type"] == "road_works"
result = decide(dict(event.data), source="511", now=time.time())
assert result.broadcast is False
assert result.lifecycle == "suppress"
def test_crash_event_still_broadcasts_through_incident_gate_end_to_end(adapter):
"""A real ITD crash payload, parsed and translated by the adapter,
still broadcasts (gating layer is unaffected for non-work-zone sub_types)."""
from meshai.notifications.gating.incident import decide
raw = _itd_raw(
event_id="INC-7002",
event_type="accidentsAndIncidents",
event_sub_type="crash",
description="Vehicle collision, right lane blocked",
is_full_closure=False,
)
event = _through_pipeline(adapter, raw)
assert event is not None
assert event.data["sub_type"] == "incident"
# ============================================================
# FULL-CLOSURE vs. PARTIAL-RESTRICTION LANGUAGE TESTS
#
# The loose `is_closure` property (also used for severity/summary) matches
# any "closed" substring, including partial-restriction wording like "All
# Shoulders Closed" that ITD explicitly does not want broadcast. The
# sub_type mapping's full-closure branch must use a stricter,
# mapping-only determination instead.
# ============================================================
def test_all_shoulders_closed_roadwork_is_road_works(adapter):
"""'All Shoulders Closed' is a partial restriction, not a full closure
-> stays 'road_works' (suppressed), despite containing 'closed'."""
raw = _itd_raw(
event_id="RW-8001",
event_type="roadwork",
description="All Shoulders Closed",
is_full_closure=False,
)
event = _through_pipeline(adapter, raw)
assert event is not None
assert event.data["sub_type"] == "road_works"
@pytest.mark.parametrize(
"description",
["One lane closed", "Right lane closed", "Left lane closed"],
)
def test_single_lane_closed_roadwork_is_road_works(adapter, description):
"""Single-lane restriction wording -> stays 'road_works' (suppressed)."""
raw = _itd_raw(
event_id="RW-8002",
event_type="roadwork",
description=description,
is_full_closure=False,
)
event = _through_pipeline(adapter, raw)
assert event is not None
assert event.data["sub_type"] == "road_works"
def test_ramp_closed_roadwork_is_road_works(adapter):
"""'Ramp closed' is a partial restriction -> stays 'road_works' (suppressed)."""
raw = _itd_raw(
event_id="RW-8003",
event_type="roadwork",
description="Ramp closed for repaving",
is_full_closure=False,
)
event = _through_pipeline(adapter, raw)
assert event is not None
assert event.data["sub_type"] == "road_works"
def test_all_lanes_closed_roadwork_is_road_closed(adapter):
"""'All lanes closed' is a genuine full closure -> 'road_closed', and
must STILL BROADCAST (this is the safety case: a real closure phrased
as roadwork must not be suppressed by the work-zone gate)."""
from meshai.notifications.gating.incident import decide
raw = _itd_raw(
event_id="RW-8004",
event_type="roadwork",
description="All lanes closed for bridge demolition",
is_full_closure=False,
)
event = _through_pipeline(adapter, raw)
assert event is not None
assert event.data["sub_type"] == "road_closed"
result = decide(dict(event.data), source="511", now=time.time())
assert result.broadcast is True
assert result.lifecycle != "suppress"
def test_is_full_closure_flag_true_no_closure_words_is_road_closed(adapter):
"""IsFullClosure=true + roadwork, with a description containing no
closure wording at all -> still 'road_closed' via the raw flag."""
raw = _itd_raw(
event_id="RW-8005",
event_type="roadwork",
description="Bridge deck replacement in progress",
is_full_closure=True,
)
event = _through_pipeline(adapter, raw)
assert event is not None
assert event.data["sub_type"] == "road_closed"
def test_road_closed_due_to_rock_slide_incident_is_road_closed(adapter):
"""Explicit full-closure wording on an accidentsAndIncidents event
still wins the mapping -> 'road_closed'."""
raw = _itd_raw(
event_id="INC-8006",
event_type="accidentsAndIncidents",
event_sub_type="hazard",
cause="Rock slide",
description="Road closed due to rock slide",
is_full_closure=False,
)
event = _through_pipeline(adapter, raw)
assert event is not None
assert event.data["sub_type"] == "road_closed"
def test_lane_restrictions_roadwork_is_road_works(adapter):
"""'Lane restrictions' wording -> stays 'road_works' (suppressed)."""
raw = _itd_raw(
event_id="RW-8007",
event_type="roadwork",
description="Lane restrictions in effect through Friday",
is_full_closure=False,
)
event = _through_pipeline(adapter, raw)
assert event is not None
assert event.data["sub_type"] == "road_works"
def test_alternating_traffic_roadwork_is_road_works(adapter):
"""'Alternating' one-lane traffic control -> stays 'road_works' (suppressed)."""
raw = _itd_raw(
event_id="RW-8008",
event_type="roadwork",
description="Alternating traffic controlled by flaggers",
is_full_closure=False,
)
event = _through_pipeline(adapter, raw)
assert event is not None
assert event.data["sub_type"] == "road_works"

View file

@ -200,12 +200,32 @@ def test_populates_core_fields(adapter):
assert event.source == "usgs_quake"
assert event.lat == 42.61
assert event.lon == -114.48
assert event.region == "magic_valley"
# to_event() deliberately does NOT propagate the raw dict's "region"
# (a fixed config default that never matches region_routes cell keys)
# onto the Event -- Event.region is left unset so CoverageFilter's
# geometry-based tagger (lat/lon vs named coverage areas) can set the
# real region name downstream. See test_region_left_unset_for_tagger
# and tests/test_coverage_area.py for the tagging behavior itself.
assert event.region is None
assert event.expires == evt["expires"]
assert event.timestamp == evt["quake_time"]
assert event.id
def test_region_left_unset_for_tagger(adapter):
"""Regression: to_event() must never propagate the adapter's fixed
config region (e.g. "magic_valley") onto the Event. That value never
matches a region_routes cell key ("East Idaho"/"SC Idaho"/"SW Idaho"),
so pre-setting it would silently break region-routed delivery. Leaving
event.region/regions unset lets CoverageFilter's geometry tagger (which
only fires when `not event.regions`) stamp the real area name from the
quake's actual lat/lon."""
evt = make_quake_event(lat=44.2, lon=-114.0, region="magic_valley")
event = adapter.to_event(evt)
assert event.region is None
assert event.regions == []
# ============================================================
# DEFENSIVE TESTS
# ============================================================

View file

@ -183,13 +183,16 @@ def test_api_list_towns(client):
def test_api_post_add_town(client):
# A clearly-fictitious name, not a real Idaho town -- avoids colliding
# with the town_anchors seed (real curated Idaho/neighbor towns; a
# UNIQUE(name) constraint 400s on a duplicate insert).
r = client.post("/api/town-anchors", json={
"name": "Bellevue", "lat": 43.4670, "lon": -114.2557, "state": "ID",
"name": "Testopolis", "lat": 43.4670, "lon": -114.2557, "state": "ID",
})
assert r.status_code == 200
assert r.json()["name"] == "bellevue"
assert r.json()["name"] == "testopolis"
invalidate_curation_cache()
coord = lookup_town_anchor("bellevue")
coord = lookup_town_anchor("testopolis")
assert coord is not None

View file

@ -0,0 +1,76 @@
"""Tests for meshai.notifications.evac_phase.detect_phase.
Cases are drawn from REAL FEMA IPAWS alert headline/CMAMtext/description
strings (see tests/fixtures/ipaws/) plus explicit false-positive guards, since
a wrong GO detection would broadcast an evacuation order that was never
issued.
"""
from __future__ import annotations
import pytest
from meshai.notifications.evac_phase import detect_phase
# ============================================================
# real strings -> expected phase
# ============================================================
@pytest.mark.parametrize("text, expected", [
("LEVEL 1 READY - Evacuation Status", "READY"),
("Level 2 Set Alert", "SET"),
("LEVEL I SET Alert", "SET"), # word beats numeral; err upward
("Level 3 - Go Now", "GO"),
("Level 3- Go Now", "GO"),
("Ohio Gulch GO Evacuation Status", "GO"),
("Indian Creek Set to GO.", "GO"), # highest wins, NOT SET
("Immediate Evacuation", "GO"),
("Prepare to Evacuate", "SET"),
("BRUSH FIRE", None),
("Owyhee County Closure Area", None),
("Endangered Missing Person Alert", None),
(
"Jackson County Sheriff's Office- Level 3 GO NOW evacuation notice "
"UPGRADED for JAC-126",
"GO",
),
])
def test_detect_phase_real_strings(text, expected):
assert detect_phase(text) == expected
# ============================================================
# false-positive guards — bare lowercase / narrative usage must NOT match
# ============================================================
@pytest.mark.parametrize("text", [
"residents should go to the Blaine County Fairgrounds",
"crews are set to arrive by 0600",
"evacuate via Go Creek Road",
])
def test_detect_phase_false_positive_guards(text):
assert detect_phase(text) is None
# ============================================================
# precedence + multi-arg scanning
# ============================================================
def test_precedence_highest_always_wins_regardless_of_arg_order():
# SET language in the first text, GO language in the second — GO must win.
assert detect_phase("Prepare to Evacuate", "Level 3 - Go Now") == "GO"
# Same phrases, arguments reversed — still GO (order-independent).
assert detect_phase("Level 3 - Go Now", "Prepare to Evacuate") == "GO"
def test_detect_phase_scans_all_texts_together():
# No single text alone carries a phase; combined they do (SET here,
# since "LEVEL 2" is an explicit SET phrase and no GO phrase is present).
assert detect_phase("Jackson County Sheriff's Office", "Level 2 Set Alert") == "SET"
def test_detect_phase_none_texts_and_empty_input_are_safe():
assert detect_phase(None, None) is None
assert detect_phase() is None
assert detect_phase("") is None
assert detect_phase(None, "BRUSH FIRE", None) is None

View file

@ -217,7 +217,10 @@ class TestFormatterGolden:
# Golden literal (captured from the live fire_format all-clear branch;
# this is the SAME format string handle_wfigs used to build inline
# before its removal -- see notifications/formatters/fire.py::_render_allclear).
assert new_wire == "✅ Cache Peak Fire — contained & closed\n1,847 ac | 23% contained | 24 mi S of Burley"
# "oakley" is nearer to 42.197,-113.710 than "burley" and is now in
# the town_anchors seed after the seed-list sync (Fix 2), so it wins
# the anchor resolution instead of the previously-nearest seeded town.
assert new_wire == "✅ Cache Peak Fire — contained & closed\n1,847 ac | 23% contained | 9 mi E of Oakley"
# ─────────────────────────────────────────────────────────────────────────────
@ -323,6 +326,102 @@ class TestGateSequenceParity:
assert gate.lifecycle == "suppress"
# ─────────────────────────────────────────────────────────────────────────────
# 2b. Tombstone mute — decide() must honor fires.tombstoned_at
# ─────────────────────────────────────────────────────────────────────────────
class TestTombstoneMute:
"""A non-null fires.tombstoned_at is a durable operator kill-switch: an
incident row stamped tombstoned must never broadcast again through
decide() -- new, update, or otherwise -- regardless of forward
acreage/containment growth or the 8h cooldown. This is the SAME column
the dashboard's /api/env/active view already filters on
(WHERE tombstoned_at IS NULL); decide() previously never read it."""
_IRWIN = "IRWIN-TOMBSTONE-1"
def _seed(self, conn, *, acres, contained, last_bcast_at, tombstoned_at=None):
_write_fire_state(
conn, irwin_id=self._IRWIN, name="Lake Channel Fire",
acres=acres, contained_pct=contained, lat=42.0, lon=-114.0,
county="Twin Falls", state="ID", now=int(last_bcast_at or 1_000_000))
conn.execute(
"UPDATE fires SET last_broadcast_at=?, last_broadcast_acres=?, "
"last_broadcast_contained=?, tombstoned_at=? WHERE irwin_id=?",
(last_bcast_at, acres, contained, tombstoned_at, self._IRWIN),
)
def _incident(self, *, acres, contained):
return {
"_kind": "wfigs_incident", "irwin_id": self._IRWIN,
"incident_name": "Lake Channel Fire", "acres": acres,
"contained_pct": contained, "declared_at_epoch": None,
"lat": 42.0, "lon": -114.0, "county": "Twin Falls", "state": "ID",
}
# (a) non-tombstoned row, forward growth past cooldown -> still broadcasts.
def test_non_tombstoned_growth_still_broadcasts(self, mem_db):
self._seed(mem_db, acres=100, contained=10,
last_bcast_at=1_000_000 - 9 * 3600, tombstoned_at=None)
gate = fire_decide(self._incident(acres=200, contained=10),
source="wfigs", now=1_000_000.0)
assert gate.broadcast is True
assert gate.lifecycle == "update"
# (b) same row, tombstoned_at set -> suppressed regardless of growth/cooldown.
def test_tombstoned_row_suppressed_despite_growth(self, mem_db):
self._seed(mem_db, acres=100, contained=10,
last_bcast_at=1_000_000 - 9 * 3600, tombstoned_at=999_000)
gate = fire_decide(self._incident(acres=9999, contained=99),
source="wfigs", now=1_000_000.0)
assert gate.broadcast is False
assert gate.lifecycle == "suppress"
assert "tombston" in gate.reason.lower()
# (c) brand-new incident, not tombstoned (no row at all) -> broadcasts new.
def test_brand_new_incident_not_tombstoned_broadcasts(self, mem_db):
gate = fire_decide(self._incident(acres=50, contained=0),
source="wfigs", now=1_000_000.0)
assert gate.broadcast is True
assert gate.lifecycle == "new"
# (d) defensive: a row lacking the tombstoned_at column must not raise.
def test_missing_tombstoned_at_column_does_not_raise(self, mem_db, monkeypatch):
self._seed(mem_db, acres=100, contained=10,
last_bcast_at=1_000_000 - 9 * 3600, tombstoned_at=None)
import sqlite3
import meshai.notifications.gating.fire as fire_mod
class _NoTombstoneColConn:
"""Proxies the real connection but simulates a pre-v12 schema:
any SELECT naming tombstoned_at raises OperationalError, exactly
as a real un-migrated `fires` table would."""
def __init__(self, real):
self._real = real
def execute(self, sql, *args, **kwargs):
if "tombstoned_at" in sql and "SELECT" in sql.upper():
raise sqlite3.OperationalError(
"no such column: tombstoned_at")
return self._real.execute(sql, *args, **kwargs)
def __getattr__(self, name):
return getattr(self._real, name)
monkeypatch.setattr(
fire_mod, "get_db", lambda: _NoTombstoneColConn(mem_db))
gate = fire_decide(self._incident(acres=200, contained=10),
source="wfigs", now=1_000_000.0)
# Falls back to "not tombstoned" and proceeds with normal gating --
# must not raise, and growth past cooldown still broadcasts.
assert gate.broadcast is True
assert gate.lifecycle == "update"
# ─────────────────────────────────────────────────────────────────────────────
# 3. Registration — the three explicit categories resolve; FIRMS does not
# ─────────────────────────────────────────────────────────────────────────────

View file

@ -8,6 +8,7 @@ Ported-behavior coverage:
* geometry-path Point -> centroid extraction
"""
from __future__ import annotations
import asyncio
import json
@ -203,7 +204,7 @@ def test_cold_start_silent_first_poll_seeds_persists_no_emit():
# Stub the network fetch with one active outage.
adapter._fetch = lambda url, extra_headers=None: _payload([IDAHO_POWER_ITEM])
store.refresh() # poll 1 == pre-existing backlog
asyncio.run(store.refresh()) # poll 1 == pre-existing backlog
# Nothing broadcast on the cold-start poll...
assert captured == [], "first poll must broadcast NOTHING (cold-start seed)"
@ -220,14 +221,14 @@ def test_cold_start_silent_first_poll_seeds_persists_no_emit():
def test_later_poll_broadcasts_newly_received_item():
store, adapter, captured = _make_store_with_generic()
adapter._fetch = lambda url, extra_headers=None: _payload([IDAHO_POWER_ITEM])
store.refresh() # poll 1 — seed silently
asyncio.run(store.refresh()) # poll 1 — seed silently
assert captured == []
# A genuinely NEW outage appears on a later poll -> it must broadcast.
new_item = dict(IDAHO_POWER_ITEM, omsOutageId="456", omsCustomerCount=99)
adapter._fetch = lambda url, extra_headers=None: _payload([IDAHO_POWER_ITEM, new_item])
adapter._last_poll.clear() # force cadence to elapse
store.refresh() # poll 2
asyncio.run(store.refresh()) # poll 2
assert len(captured) == 1, "only the newly-received outage broadcasts"
assert captured[0].category == "power_outage"
@ -367,7 +368,7 @@ def test_build_generic_detail_reader():
from meshai.notifications.env_reporter import EnvReporter
store, adapter, captured = _make_store_with_generic()
adapter._fetch = lambda url, extra_headers=None: _payload([IDAHO_POWER_ITEM])
store.refresh()
asyncio.run(store.refresh())
text = EnvReporter().build_generic_detail()
assert "idaho_power" in text

View file

@ -188,7 +188,11 @@ class TestWorkZoneGolden:
_GOLDEN = {
"0002.json": "🚧 US-91, near Chubbuck: southbound, road construction, ends Aug 17",
"0003.json": "🚧 US-95, near Wilder: southbound, ends Jul 19",
# "wilder" was added to the town_anchors seed by the seed-list sync
# (Fix 2), so the DB-anchor step now wins over the live Photon
# geocode this golden was originally captured against; the DB row's
# coords round to 1 mi S instead of Photon's sub-mile "near".
"0003.json": "🚧 US-95, 1 mi S of Wilder: southbound, ends Jul 19",
}
# Captured from the deleted normalizer's normalize() + _n_to_canonical_workzone()
@ -434,6 +438,35 @@ class TestAnchorResolve:
assert resolve_anchor(None, -116.2, max_mi=50.0) is None
assert resolve_anchor(43.6, None, max_mi=50.0) is None
def test_disabled_anchor_excluded(self, monkeypatch):
"""A disabled=0 town_anchors row must not be selected, even when it is
the closest row within max_mi the enabled flag is a hard exclude."""
import time as _time
from meshai.persistence import get_db
from meshai.notifications.formatters._anchor import resolve_anchor
from meshai import geo
# Force the Photon fallback to a known miss so a non-None result can
# only come from the (wrongly-included) disabled DB row.
monkeypatch.setattr(
geo, "nearest_town",
lambda lat, lon, max_distance_mi=50.0: None,
)
conn = get_db()
# Clear seeded anchors so only our controlled (disabled) row exists.
conn.execute("DELETE FROM town_anchors")
conn.execute(
"INSERT INTO town_anchors(name, lat, lon, state, enabled, updated_at) "
"VALUES (?, ?, ?, ?, ?, ?)",
("disabledville", -33.8688, 151.2093, "NSW", 0, _time.time()), # Sydney
)
# Event right next to the disabled row; Photon fallback is forced to
# miss → None confirms the DB step excluded the disabled row.
result = resolve_anchor(-33.870, 151.210, max_mi=50.0)
assert result is None
# ── 4. Schema conformance ────────────────────────────────────────────────────

View file

@ -0,0 +1,212 @@
"""Regression tests for the usgs_quake dispatcher-freshness + region-tagging fix.
Problem (see meshai/notifications/pipeline/dispatcher.py:359-374 and
meshai/env/usgs_quake.py): the usgs_quake adapter polls a rolling PAST-DAY
USGS feed (2.5_day.geojson), so a genuinely first-seen quake is routinely
already 10min-20h old by the time it's detected. The generic per-toggle
freshness_seconds (600s) silently dropped essentially every native quake
before broadcast -- confirmed against production: quake_events had 12
first-sighting rows (decide() ran, magnitude/region gate passed, DB row
inserted) with last_broadcast_at=NULL on every single one (commit() never
reached because the staleness filter dropped the event downstream first).
Fix 1 (dispatcher.py): earthquake_event gets its own adapter_config-backed
freshness override (adapter_config.usgs_quake.freshness_seconds, default
3600s), mirroring the existing wfigs/"fire" override -- but scoped to the
CATEGORY, not the "seismic" family/toggle, because stream_flood_warning /
stream_high_water (hydro) also live under toggle="seismic" and must keep
using the generic per-toggle freshness unchanged.
Fix 2 (env/usgs_quake.py): to_event() no longer passes the adapter's fixed
config region ("magic_valley", which never matches a region_routes cell
key) onto the Event. Event.region is left unset so CoverageFilter's
geometry-based tagger (lat/lon vs named coverage areas) can stamp the real
region name.
"""
from __future__ import annotations
import asyncio
import time
from unittest.mock import MagicMock
import pytest
from meshai.config import Config
from meshai.coverage_area import MonitoringArea
from meshai.env.usgs_quake import USGSQuakeAdapter
from meshai.notifications.events import make_event
from meshai.notifications.pipeline.coverage_filter import CoverageFilter
from meshai.notifications.pipeline.dispatcher import Dispatcher
# --------------------------------------------------------------------------
# Shared dispatcher-test helpers (mirror tests/test_v052_dispatcher.py)
# --------------------------------------------------------------------------
class RecChannel:
def __init__(self, rec):
self.rec = rec
async def deliver(self, payload, rule):
self.rec.append({"name": rule.name, "message": payload.message})
return True
def _make_dispatcher(cfg):
rec: list = []
d = Dispatcher(cfg, lambda rule, conn: RecChannel(rec), connector=None)
return d, rec
def _dispatch_one(cfg, event):
d, rec = _make_dispatcher(cfg)
asyncio.run(d.dispatch(event))
return d, rec
def _quake_cfg():
"""Config with the seismic toggle enabled, cold-start grace disabled."""
cfg = Config()
cfg.notifications.rules = []
cfg.notifications.cold_start_grace_seconds = 0
t = cfg.notifications.toggles["seismic"]
t.enabled = True
t.min_severity = "routine"
t.regions = []
t.severity_channels = {
"routine": ["mesh_broadcast"],
"priority": ["mesh_broadcast"],
"immediate": ["mesh_broadcast"],
}
t.cooldown_seconds = 0
# Generic per-toggle freshness -- deliberately tight (600s, the old
# default) so these tests prove the earthquake_event category is NOT
# using this value anymore.
t.freshness_seconds = 600
return cfg
def _quake_event(age_seconds: float, event_id="us6000abcd", lat=42.6, lon=-114.5):
return make_event(
source="usgs_quake",
category="earthquake_event",
severity="routine",
title=f"M3.0 -- {event_id}",
timestamp=time.time() - age_seconds,
lat=lat,
lon=lon,
group_key=event_id,
inhibit_keys=[event_id],
)
# ============================================================
# (a) fresh quake passes under the new per-adapter override
# ============================================================
def test_fresh_quake_passes_staleness_gate_with_override():
"""A 30-minute-old quake (1800s) exceeds the generic 600s toggle window
but must pass under the adapter_config.usgs_quake.freshness_seconds
override (default 3600s)."""
cfg = _quake_cfg()
event = _quake_event(age_seconds=1800)
d, rec = _dispatch_one(cfg, event)
assert len(rec) == 1, "30-min-old quake must broadcast under the 3600s override"
assert d.dispatch_stats()["stale_dropped"] == 0
# ============================================================
# (b) old quake (12h) is still dropped
# ============================================================
def test_old_quake_still_dropped_by_override():
"""A 12-hour-old quake (43200s) must still be dropped -- the override
widens the window, it does not disable the staleness check."""
cfg = _quake_cfg()
event = _quake_event(age_seconds=12 * 3600)
d, rec = _dispatch_one(cfg, event)
assert rec == [], "12h-old quake must still be dropped as stale"
assert d.dispatch_stats()["stale_dropped"] == 1
def test_hydro_seismic_sibling_unaffected_by_quake_override():
"""Guardrail: stream_flood_warning shares toggle='seismic' with
earthquake_event but must keep using the GENERIC per-toggle freshness
(600s here), not the quake-specific 3600s override. A 1800s-old hydro
event must still be dropped."""
cfg = _quake_cfg()
event = make_event(
source="usgs", category="stream_flood_warning", severity="priority",
title="Snake River nr Twin Falls 12.8 ft",
timestamp=time.time() - 1800,
)
d, rec = _dispatch_one(cfg, event)
assert rec == [], "hydro sibling must NOT inherit the quake freshness override"
assert d.dispatch_stats()["stale_dropped"] == 1
# ============================================================
# (c) region-tagging from real lat/lon
# ============================================================
SW_IDAHO = MonitoringArea(name="SW Idaho", west=-117.993408, south=42.05,
east=-115.389404, north=44.331707)
SC_IDAHO = MonitoringArea(name="SC Idaho", west=-115.389404, south=42.05,
east=-112.8, north=44.331707)
EAST_IDAHO = MonitoringArea(name="East Idaho", west=-112.8, south=41.9,
east=-110.9, north=45.331707)
PROD_AREAS = [SW_IDAHO, SC_IDAHO, EAST_IDAHO]
def _quake_adapter():
cfg = MagicMock()
cfg.feed_url = "https://example.test/feed.geojson"
cfg.min_magnitude = 2.5
cfg.bbox = [-115.5, 42.0, -110.0, 45.2]
cfg.region = "magic_valley"
cfg.tick_seconds = 300
return USGSQuakeAdapter(cfg)
def test_quake_event_region_tags_to_matching_coverage_area():
"""A quake with real Idaho lat/lon, run through the actual adapter's
to_event() and then the real CoverageFilter (named exactly like prod's
SW/SC/East Idaho areas), must end up tagged with a region name that
matches a region_routes cell key -- NOT the adapter's stale
"magic_valley" default."""
adapter = _quake_adapter()
raw = {
"source": "usgs_quake",
"event_id": "us7000example",
"event_type": "Earthquake",
"severity": "routine",
"headline": "M2.7 -- 10 km N of Twin Falls, ID",
"magnitude": 2.7,
"place": "10 km N of Twin Falls, ID",
"depth_km": 8.0,
"sig": 80,
"url": "https://earthquake.usgs.gov/x",
"region": "magic_valley",
"lat": 42.61, # Twin Falls -- falls inside SC Idaho
"lon": -114.48,
"quake_time": time.time(),
"expires": time.time() + 86400,
"fetched_at": time.time(),
}
event = adapter.to_event(raw)
assert event is not None
assert event.region is None # adapter no longer pre-stamps a region
received: list = []
flt = CoverageFilter(next_handler=received.append, areas=PROD_AREAS, enabled=True)
flt.handle(event)
assert len(received) == 1, "in-region quake must pass the coverage gate"
tagged = received[0]
assert tagged.region == "SC Idaho", (
f"expected geometry tagging to set 'SC Idaho', got {tagged.region!r} "
"(pre-fix this would have stayed 'magic_valley' and never matched a "
"region_routes cell key)"
)
assert tagged.regions == ["SC Idaho"]

View file

@ -354,3 +354,76 @@ def test_parse_norad_ids_handles_comma_string():
assert SatpassAdapter._parse_norad_ids("25544, 33591") == [25544, 33591]
assert SatpassAdapter._parse_norad_ids([25544, "33591"]) == [25544, 33591]
assert SatpassAdapter._parse_norad_ids([]) == []
# ══════════════════════════════════════════════════════════════════════
# 8. observer_list FLOWS INTO THE GATE'S data DICT (region-tagging fix)
# ══════════════════════════════════════════════════════════════════════
#
# gate_consolidated_pass() builds the event `data` dict that rides all the
# way to the dispatcher via to_event(). Satpass events carry no lat/lon/
# geometry, so coverage_area.observer_region_names() reading
# event.data["observer_list"] is the ONLY path that can region-tag a
# satpass event for the region_routes matrix. These tests pin that the
# gate actually populates it (and stays safe when it can't).
def test_gate_consolidated_pass_data_contains_observer_list():
"""The (wire, data) tuple returned by gate_consolidated_pass() must
carry the same comma-joined observer_list already written to the
satpass_events audit column."""
_enable_satpass_db(dry_run=False)
from meshai.env.satellite import pass_format as sh
consolidated = {
"consolidated_id": "25544:900000",
"norad_id": 25544,
"sat_name": "ISS (ZARYA)",
"max_elevation": 70.0,
"aos_epoch": 1_000_000,
"los_epoch": 1_000_300,
"aos_compass": "SW",
"los_compass": "NE",
"peak_compass": "S",
"entry_observer": "Boise",
"exit_observer": "Twin Falls",
"observer_list": "boise,twin",
}
result = sh.gate_consolidated_pass(consolidated, now=0)
assert result is not None
_, data = result
assert data["observer_list"] == "boise,twin"
def test_gate_consolidated_pass_missing_observer_list_is_defensive():
"""A `consolidated` dict with no observer_list must not raise, and the
resulting data["observer_list"] must be falsy (never a garbage value
that would resolve to a bogus region)."""
_enable_satpass_db(dry_run=False)
from meshai.env.satellite import pass_format as sh
consolidated = {
"consolidated_id": "25544:900001",
"norad_id": 25544,
"sat_name": "ISS (ZARYA)",
"max_elevation": 40.0,
"aos_epoch": 2_000_000,
"los_epoch": 2_000_300,
"aos_compass": "SW",
"los_compass": "NE",
"peak_compass": "S",
# entry_observer / exit_observer / observer_list all deliberately absent
}
result = sh.gate_consolidated_pass(consolidated, now=0)
assert result is not None
_, data = result
assert not data.get("observer_list")
# And observer_region_names() must treat that as "no region", not raise.
from meshai.coverage_area import MonitoringArea, observer_region_names
from meshai.notifications.events import make_event
event = make_event(source="satpass", category="sat_pass", severity="routine",
title="Pass", data=data)
areas = [MonitoringArea(north=44.0, south=42.0, east=-113.0, west=-117.0,
name="SW Idaho")]
assert observer_region_names(event, areas) == []

View file

@ -245,3 +245,111 @@ class TestSatpassToggle:
def test_categories_for_toggle_satpass_returns_sat_pass(self):
result = categories_for_toggle("satpass")
assert result == ["sat_pass"]
# ---------------------------------------------------------------------------
# End-to-end: a satpass Event built through the REAL native-adapter path
# (predict -> consolidate -> gate_consolidated_pass -> to_event) must
# region-tag via event_region_names(), using the two production observers.
# This is the regression test for the observer_list wiring bug: before the
# fix, gate_consolidated_pass()'s data dict never carried observer_list, so
# this resolved to [] no matter how the areas were configured.
# ---------------------------------------------------------------------------
# Production ground stations (see adapter_config observers): Treasure Valley
# (Boise area) and Magic Valley (Twin Falls area).
_TREASURE_VALLEY = {"slug": "treasure_valley", "name": "Treasure Valley",
"lat": 43.6, "lon": -116.2, "alt_m": 0.0}
_MAGIC_VALLEY = {"slug": "magic_valley", "name": "Magic Valley",
"lat": 42.5558, "lon": -114.4701, "alt_m": 0.0}
# Coarse named boxes that cover each observer, standing in for the real
# region_routes cells ("SW Idaho" covers the Treasure Valley/Boise area,
# "SC Idaho" covers the Magic Valley/Twin Falls area).
_SW_IDAHO = MonitoringArea(north=44.5, south=42.8, east=-115.0, west=-117.5,
name="SW Idaho")
_SC_IDAHO_PROD = MonitoringArea(north=43.2, south=42.0, east=-113.0, west=-115.2,
name="SC Idaho")
_PROD_AREAS = [_SW_IDAHO, _SC_IDAHO_PROD]
class TestSatpassEndToEndRegionTagging:
def test_real_gate_path_region_tags_nonempty(self, monkeypatch):
"""Drive the actual SatpassAdapter tick -> gate -> to_event path
(no shortcuts through gate_consolidated_pass or make_event) with the
two production observer coordinates, then confirm event_region_names
returns a non-empty list built from event.data['observer_list']."""
import json as _json
from datetime import datetime, timezone
from meshai.adapter_config import invalidate_cache
from meshai.config import SatpassConfig
from meshai.env.satellite.pass_predictor import PassInfo
from meshai.env.satellite.tle_store import upsert_tle
from meshai.env.satpass import SatpassAdapter
from meshai.persistence import get_db
# -- seed a fresh ISS TLE --------------------------------------------------
conn = get_db()
fresh = datetime.now(timezone.utc).isoformat()
iss_l1 = "1 25544U 98067A 26182.50000000 .00016717 00000-0 10270-3 0 9008"
iss_l2 = "2 25544 51.6400 208.9163 0007417 17.6777 85.6621 15.54225995 12345"
upsert_tle(conn, 25544, "ISS (ZARYA)", iss_l1, iss_l2, fresh)
# -- enable satpass, dry_run off so the gate actually returns data --------
conn.execute("UPDATE adapter_config SET value_json='true' "
"WHERE adapter='satpass' AND key='enabled'")
conn.execute("UPDATE adapter_config SET value_json='false' "
"WHERE adapter='satpass' AND key='dry_run'")
conn.execute("UPDATE adapter_config SET value_json=? "
"WHERE adapter='satpass' AND key='max_broadcasts_per_hour'",
(_json.dumps(100),))
invalidate_cache()
# -- patch observers + predictor -------------------------------------------
monkeypatch.setattr(
"meshai.persistence.observer_locations.get_observers",
lambda *a, **k: [_TREASURE_VALLEY, _MAGIC_VALLEY])
t0 = (1783000000 // 3600) * 3600 + 100
def _fake_compute_passes(l1, l2, lat, lon, alt, window_h, min_el, now):
def _pi(aos, los, max_el, az_aos, az_los, az_peak):
peak = (aos + los) // 2
return PassInfo(
aos_time=datetime.fromtimestamp(aos, tz=timezone.utc),
los_time=datetime.fromtimestamp(los, tz=timezone.utc),
peak_time=datetime.fromtimestamp(peak, tz=timezone.utc),
max_elevation=max_el,
azimuth_at_aos=az_aos, azimuth_at_los=az_los,
azimuth_at_peak=az_peak)
if abs(lat - _TREASURE_VALLEY["lat"]) < 0.1:
return [_pi(t0, t0 + 300, 40.0, 225, 300, 90)]
if abs(lat - _MAGIC_VALLEY["lat"]) < 0.1:
return [_pi(t0 + 60, t0 + 400, 70.0, 270, 45, 180)]
return []
monkeypatch.setattr(
"meshai.env.satellite.pass_predictor.compute_passes",
_fake_compute_passes)
cfg = SatpassConfig(enabled=True, feed_source="native",
norad_ids=[25544], min_elevation_deg=10.0,
window_hours=24)
adapter = SatpassAdapter(cfg)
assert adapter.tick(now=t0 - 1800) is True # AOS 30 min out -> imminent
staged = adapter.get_events()
assert len(staged) == 1
event = adapter.to_event(staged[0])
assert event is not None
# The bug: before the fix, data["observer_list"] was never set, so
# this was always [] regardless of area config.
assert event.data.get("observer_list"), (
"gate_consolidated_pass() did not populate observer_list on the "
"event data dict")
names = event_region_names(event, _PROD_AREAS)
assert names, "satpass event failed to region-tag via observer_list"
assert set(names) == {"SW Idaho", "SC Idaho"}

View file

@ -14,6 +14,7 @@ These tests drive the real EnvironmentalStore + EventBus with a fake adapter
whose per-poll batch we control, and assert exactly which events reach the bus.
"""
from __future__ import annotations
import asyncio
from meshai.env.store import EnvironmentalStore, _key_ext
from meshai.config import EnvironmentalConfig
@ -79,7 +80,7 @@ def test_first_poll_seeds_and_broadcasts_nothing():
store, adapter, captured = _make_store()
adapter.set_batch(["A", "B", "C"])
store.refresh() # poll 1 — the backlog
asyncio.run(store.refresh()) # poll 1 — the backlog
assert captured == [], "first poll must broadcast NOTHING (backlog seed)"
@ -88,11 +89,11 @@ def test_second_poll_emits_only_newly_received():
store, adapter, captured = _make_store()
adapter.set_batch(["A", "B", "C"])
store.refresh() # poll 1: seed
asyncio.run(store.refresh()) # poll 1: seed
assert _emitted_ids(captured) == []
adapter.set_batch(["A", "B", "C", "D"])
store.refresh() # poll 2: only D is new
asyncio.run(store.refresh()) # poll 2: only D is new
assert _emitted_ids(captured) == ["D"]
@ -100,11 +101,11 @@ def test_unchanged_poll_emits_nothing():
store, adapter, captured = _make_store()
adapter.set_batch(["A", "B", "C"])
store.refresh() # poll 1: seed
asyncio.run(store.refresh()) # poll 1: seed
adapter.set_batch(["A", "B", "C", "D"])
store.refresh() # poll 2: D
asyncio.run(store.refresh()) # poll 2: D
adapter.set_batch(["A", "B", "C", "D"])
store.refresh() # poll 3: nothing new
asyncio.run(store.refresh()) # poll 3: nothing new
assert _emitted_ids(captured) == ["D"], "poll 3 has no new items"
@ -113,21 +114,21 @@ def test_restart_reseeds_and_never_rebroadcasts_backlog():
# Process 1 sees A,B,C,D and broadcasts D.
store1, adapter1, cap1 = _make_store()
adapter1.set_batch(["A", "B", "C"])
store1.refresh()
asyncio.run(store1.refresh())
adapter1.set_batch(["A", "B", "C", "D"])
store1.refresh()
asyncio.run(store1.refresh())
assert _emitted_ids(cap1) == ["D"]
# RESTART: a fresh store has an empty seen-set. The SAME backlog [A,B,C,D]
# arriving on its first poll must be re-seeded silently, not re-broadcast.
store2, adapter2, cap2 = _make_store()
adapter2.set_batch(["A", "B", "C", "D"])
store2.refresh()
asyncio.run(store2.refresh())
assert cap2 == [], "restart must NEVER re-broadcast the existing backlog"
# And a genuinely new item after the restart still broadcasts once.
adapter2.set_batch(["A", "B", "C", "D", "E"])
store2.refresh()
asyncio.run(store2.refresh())
assert _emitted_ids(cap2) == ["E"]
@ -135,9 +136,9 @@ def test_stable_key_prevents_reemit_when_batch_reorders():
# The same real-world items in a different order are NOT "newly received".
store, adapter, captured = _make_store()
adapter.set_batch(["A", "B", "C"])
store.refresh() # seed
asyncio.run(store.refresh()) # seed
adapter.set_batch(["C", "A", "B"]) # reordered, same items
store.refresh()
asyncio.run(store.refresh())
assert captured == [], "reordering the same items emits nothing"
@ -147,12 +148,12 @@ def test_disabled_for_days_then_backlog_is_not_broadcast():
store, adapter, captured = _make_store()
backlog = [f"evt{i}" for i in range(200)]
adapter.set_batch(backlog)
store.refresh() # first poll after re-enable
asyncio.run(store.refresh()) # first poll after re-enable
assert captured == [], "a days-old backlog is seeded silently, never sent"
# Only a truly new arrival afterward is announced.
adapter.set_batch(backlog + ["fresh"])
store.refresh()
asyncio.run(store.refresh())
assert _emitted_ids(captured) == ["fresh"]
@ -290,11 +291,11 @@ def test_persistent_preseed_known_suppressed_new_emitted():
assert len(store._seen["wzdx"]) == 5
adapter.set_batch(known)
store.refresh()
asyncio.run(store.refresh())
assert captured == [], "all 5 are durably-known → zero broadcast"
adapter.set_batch(known + ["z_new"])
store.refresh()
asyncio.run(store.refresh())
assert _emitted_ids(captured) == ["z_new"], "only the not-in-table id broadcasts"
@ -307,9 +308,9 @@ def test_persistent_preseed_cross_tick_staging_no_leak():
store, captured = _build_store(_GENERIC_NAME, adapter)
adapter.set_batch(["A"])
store.refresh() # tick 1: only A present
asyncio.run(store.refresh()) # tick 1: only A present
adapter.set_batch(["A", "B"])
store.refresh() # tick 2: B appears (backlog)
asyncio.run(store.refresh()) # tick 2: B appears (backlog)
assert captured == [], "B is durably-known — must NOT leak on a later tick"
# CONTROL: identical staging but NO durable rows → B leaks (proves the
@ -325,11 +326,11 @@ def test_persistent_preseed_cross_tick_staging_no_leak():
# Re-point ctrl events to a fresh source with no durable rows.
for e in ctrl._batch:
e["source"] = "wzdx_ctrl"
store2.refresh()
asyncio.run(store2.refresh())
ctrl.set_batch(["A", "B"])
for e in ctrl._batch:
e["source"] = "wzdx_ctrl"
store2.refresh()
asyncio.run(store2.refresh())
assert [e.title for e in cap2] == ["B"], "without a durable record, B leaks"
@ -343,11 +344,11 @@ def test_incremental_empty_first_tick_then_only_new_broadcasts():
store, captured = _build_store(_GENERIC_NAME, adapter)
adapter.set_batch([]) # empty first tick
store.refresh()
asyncio.run(store.refresh())
assert captured == [], "empty tick emits nothing"
adapter.set_batch(["A", "B", "C"]) # backlog A,B + new C
store.refresh()
asyncio.run(store.refresh())
assert _emitted_ids(captured) == ["C"], "only the never-received C broadcasts"
@ -360,18 +361,18 @@ def test_restart_against_same_persistent_db_never_rebroadcasts():
a1 = _FakeWZDx()
store1, cap1 = _build_store(_GENERIC_NAME, a1)
a1.set_batch(backlog)
store1.refresh()
asyncio.run(store1.refresh())
assert cap1 == [], "process 1: durable backlog is silent"
# RESTART: brand-new store, same persistent DB → pre-seed reloads.
a2 = _FakeWZDx()
store2, cap2 = _build_store(_GENERIC_NAME, a2)
a2.set_batch(backlog)
store2.refresh()
asyncio.run(store2.refresh())
assert cap2 == [], "restart must NEVER re-broadcast the durable backlog"
a2.set_batch(backlog + ["E"])
store2.refresh()
asyncio.run(store2.refresh())
assert _emitted_ids(cap2) == ["E"], "a genuinely-new item still broadcasts once"
@ -385,11 +386,11 @@ def test_persistent_preseed_quake_by_event_id():
assert len(store._seen["usgs_quake"]) == 2
adapter.set_batch(["us1000aaaa", "us1000bbbb"])
store.refresh()
asyncio.run(store.refresh())
assert captured == [], "both quakes already received → zero broadcast"
adapter.set_batch(["us1000aaaa", "us1000bbbb", "us1000cccc"])
store.refresh()
asyncio.run(store.refresh())
assert _emitted_ids(captured) == ["us1000cccc"], "only the new quake broadcasts"
@ -402,11 +403,11 @@ def test_no_durable_rows_falls_back_to_silent_first_poll():
assert "wzdx" not in store._seeded, "0 durable rows → not pre-marked seeded"
adapter.set_batch(["A", "B"])
store.refresh()
asyncio.run(store.refresh())
assert captured == [], "first non-empty poll on a fresh DB is silent"
adapter.set_batch(["A", "B", "C"])
store.refresh()
asyncio.run(store.refresh())
assert _emitted_ids(captured) == ["C"]
@ -486,9 +487,9 @@ def test_persistent_preseed_roads511_by_external_id():
assert len(store._seen["511"]) == 4
adapter.set_batch(known)
store.refresh()
asyncio.run(store.refresh())
assert captured == [], "all 4 durably-known 511 rows → zero broadcast"
adapter.set_batch(known + ["511_99"])
store.refresh()
asyncio.run(store.refresh())
assert _emitted_ids(captured) == ["511_99"], "only the not-in-table id broadcasts"

View file

@ -20,6 +20,7 @@ adapter whose per-poll coalesced set we control, then assert directly against
traffic_events AND against the bus (nothing must ever be dispatched).
"""
from __future__ import annotations
import asyncio
from meshai.env.store import EnvironmentalStore
from meshai.config import EnvironmentalConfig
@ -147,7 +148,7 @@ def test_first_poll_persists_current_set_and_broadcasts_nothing():
store, captured = _build_store(adapter)
adapter.set_zones(ZONES3)
store.refresh() # first (cold-start) poll
asyncio.run(store.refresh()) # first (cold-start) poll
rows = _wzdx_rows()
exts = {r["external_id"] for r in rows}
@ -171,7 +172,7 @@ def test_columns_match_summary_and_dm_queries():
adapter = _FakeWZDx()
store, _ = _build_store(adapter)
adapter.set_zones([ZONES3[1]]) # the full_closure I-84 zone
store.refresh()
asyncio.run(store.refresh())
r = _wzdx_rows()[0]
assert r["road"] == "I-84"
@ -188,12 +189,12 @@ def test_subsequent_poll_reconciles_removed_zone():
adapter = _FakeWZDx()
store, captured = _build_store(adapter)
adapter.set_zones(ZONES3)
store.refresh()
asyncio.run(store.refresh())
assert len(_wzdx_rows()) == 3
# Next poll: US-20 dropped out; I-84 + ID-55 remain.
adapter.set_zones([ZONES3[1], ZONES3[2]])
store.refresh()
asyncio.run(store.refresh())
exts = {r["external_id"] for r in _wzdx_rows()}
assert exts == {ZONES3[1]["ext"], ZONES3[2]["ext"]}, (
@ -207,11 +208,11 @@ def test_empty_or_failed_fetch_does_not_wipe_existing_rows():
adapter = _FakeWZDx()
store, _ = _build_store(adapter)
adapter.set_zones(ZONES3)
store.refresh()
asyncio.run(store.refresh())
assert len(_wzdx_rows()) == 3
adapter.set_raw([]) # empty/failed poll
store.refresh()
asyncio.run(store.refresh())
assert len(_wzdx_rows()) == 3, (
"an empty fetch must NEVER wipe the existing active set")
@ -225,7 +226,7 @@ def test_upsert_preserves_first_seen_at_and_refreshes_end_at():
z = dict(ZONES3[0]); z["end_at"] = 1000
adapter.set_zones([z])
store.refresh()
asyncio.run(store.refresh())
r1 = _wzdx_rows()[0]
first_seen = r1["first_seen_at"]
assert r1["end_at"] == 1000
@ -233,7 +234,7 @@ def test_upsert_preserves_first_seen_at_and_refreshes_end_at():
# Same zone reappears with a LATER end_at.
z2 = dict(ZONES3[0]); z2["end_at"] = 5000
adapter.set_zones([z2])
store.refresh()
asyncio.run(store.refresh())
r2 = _wzdx_rows()[0]
assert r2["first_seen_at"] == first_seen, "first_seen_at must be preserved"
assert r2["end_at"] == 5000, "end_at must refresh from the feed"
@ -254,7 +255,7 @@ def test_expiry_end_at_preserved_for_not_expired_filter():
"sub_type": "x", "impact": "partial", "end_at": now - 10_000}, # expired
]
adapter.set_zones(zones)
store.refresh()
asyncio.run(store.refresh())
# All 3 persisted (ingest does not itself drop expired rows) ...
assert len(_wzdx_rows()) == 3
@ -269,7 +270,7 @@ def test_id_less_zone_is_skipped_not_fatal():
store, _ = _build_store(adapter)
good = ZONES3[0]
adapter.set_zones([good])
store.refresh()
asyncio.run(store.refresh())
assert len(_wzdx_rows()) == 1
# Poll with the good zone plus an id-less junk event.
@ -277,7 +278,7 @@ def test_id_less_zone_is_skipped_not_fatal():
junk = {"source": "wzdx", "event_id": None, "external_id": None,
"lat": 5.0, "lon": 5.0, "normalized": {}, "fetched_at": 0}
adapter._batch.append(junk)
store.refresh()
asyncio.run(store.refresh())
rows = _wzdx_rows()
assert {r["external_id"] for r in rows} == {good["ext"]}, (
@ -296,7 +297,7 @@ def test_bulk_current_set_persists_all_like_the_real_127():
for i in range(127)
]
adapter.set_zones(zones)
store.refresh()
asyncio.run(store.refresh())
assert len(_wzdx_rows()) == 127
assert _summary_visible_count(now=0) == 127, (

View file

@ -357,8 +357,10 @@ def test_k_anchor_falls_to_nearest_town(monkeypatch, mem_db):
county="Cassia")
n = _normalize_wfigs(env)
wire = _wfigs_render(n, prefix="New")
# Resolves anchor via town_anchors table (Burley @ 42.536, -113.793)
assert "Burley" in wire
# Resolves anchor via town_anchors table (Oakley @ 42.24206, -113.883058
# -- now the nearest seeded anchor to the incident's 42.197,-113.710
# after the full-list seed sync; Burley is farther away)
assert "Oakley" in wire
def test_k_anchor_falls_to_landclass(monkeypatch, mem_db):
@ -372,7 +374,8 @@ def test_k_anchor_falls_to_landclass(monkeypatch, mem_db):
n = _normalize_wfigs(env)
wire = _wfigs_render(n, prefix="New")
# Resolves nearest town from town_anchors table, overriding landclass
assert "Burley" in wire
# (Oakley is nearest to 42.197,-113.710 after the full-list seed sync)
assert "Oakley" in wire
def test_k_anchor_falls_to_county(monkeypatch, mem_db):
@ -385,7 +388,8 @@ def test_k_anchor_falls_to_county(monkeypatch, mem_db):
n = _normalize_wfigs(env)
wire = _wfigs_render(n, prefix="New")
# Resolves nearest town from town_anchors table
assert "Burley" in wire
# (Oakley is nearest to 42.197,-113.710 after the full-list seed sync)
assert "Oakley" in wire
def test_k_anchor_nearest_town_under_one_mile_says_near(monkeypatch, mem_db):
@ -398,7 +402,8 @@ def test_k_anchor_nearest_town_under_one_mile_says_near(monkeypatch, mem_db):
n = _normalize_wfigs(env)
wire = _wfigs_render(n, prefix="New")
# Anchor resolved via town_anchors; exact format depends on distance
assert "Burley" in wire
# (Oakley is nearest to 42.197,-113.710 after the full-list seed sync)
assert "Oakley" in wire
# ============================================================================