mirror of
https://github.com/zvx-echo6/meshai.git
synced 2026-06-10 17:04:45 +02:00
feat(central): v0.5.4 -- region-aware subscriptions using Central v0.9.20 regional subjects
Pre-v0.5.4 every Central subscription used a bare wildcard (central.wx.>,
central.fire.>, central.traffic.>, central.quake.>, central.hydro.>,
central.space.>), so a Magic Valley operator flipping nws -> central was
in fact subscribing to the all-US firehose and discarding 95% of events
locally. Central v0.9.20 (2026-05-28) added per-region subject suffixes
so the firehose can be filtered server-side. This wires meshai to use them.
Backend (meshai/central/consumer.py):
- New _subjects_for(adapter, region) replaces the static ADAPTER_SUBJECTS
dict. ADAPTER_SUBJECTS is retained as an alias to _SUBJECTS_BARE for any
legacy importers; the dispatcher path is unchanged.
- Per-adapter subject patterns (region='us.id' default):
nws -> central.wx.alert.us.id.> (region BEFORE wildcard)
usgs_quake -> central.quake.event.>.us.id (region AFTER wildcard)
firms -> central.fire.hotspot.>.us.id
fires -> central.fire.incident.id.> (state token at fixed depth)
central.fire.perimeter.id.>
traffic -> central.traffic.>.id (bare state, no us. prefix)
roads511 -> central.traffic.>.id (shared with traffic, sub-adapter routing)
usgs -> central.hydro.>.us.id
central.hydro.>.unknown (workaround until v0.9.20.1)
swpc -> central.space.> (planetary; region ignored)
- Empty/None region falls back to bare wildcards (pre-v0.9.20 behaviour).
- _subject_owned() pulls region from env.central.region and routes through
_subjects_for; v0.5.3 sub-adapter routing (owned-sources set) still
applies on shared subjects like central.traffic.>.id.
- start() logs the active region at connect-time for ops visibility.
Config (meshai/config.py):
- CentralConsumerConfig.region: str = "us.id". One region per consumer
applies to every central-flipped adapter; per-adapter overrides can
land in v0.6 when there is a real use case.
Frontend (dashboard-frontend/src/pages/Environment.tsx):
- Central Connection panel gets a Region text input next to URL/Durable.
- EnvConfig.central type extended with region: string.
- Static bundle rebuilt; index-DCFmSeOM.js -> index-B24tHcYj.js.
Tests:
- tests/test_central_region_routing.py (new, 9 cases): asserts the exact
v0.9.20 subject string for each adapter at region='us.id', the SWPC
global-stays-global rule, the USGS .unknown workaround, the empty-region
backward-compat fallback for all 8 adapters, and integration through
CentralConsumer._subject_owned() with the default region.
- tests/test_central_consumer.py + tests/test_central_sub_adapter_routing.py:
the two tests that asserted bare-wildcard subjects now set
env.central.region = "" explicitly to preserve their original concern
(no region semantics — backward-compat path only).
Why swpc stays global: space weather is planetary -- a CME is detected on
the sun, the geomagnetic response is hemispheric. There is no Idaho-only
solar event; subscribing per-region would only drop events we want.
Why hydro has the .unknown workaround: Central v0.9.20 leaves gauges
whose USGS state can't be inferred on central.hydro.>.unknown. Until
v0.9.20.1 backfills the state tag we subscribe to both filters to
avoid silently losing those rows. Idaho downstream-filtering on
data['_enriched']['usgs_site']['state'] is future v0.6 work.
Orthogonal to v0.5.2 dispatcher guards (staleness / cooldown / dedup)
and v0.5.3 sub-adapter routing: the region filter operates at the NATS
subscription layer (server-side), upstream of everything else.
Verified: pytest 327 passed (318 prior + 9 new region-routing tests);
py_compile clean; frontend build clean. Safe-mode preserved -- no toggle
enabled, no master enabled, no central enabled.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
ded2156024
commit
c2d5bcfbd1
8 changed files with 190 additions and 14 deletions
|
|
@ -27,7 +27,7 @@ interface EnvConfig {
|
|||
traffic: { enabled: boolean; tick_seconds: number; api_key: string; corridors: { name: string; lat: number; lon: number }[]; feed_source?: FeedSource }
|
||||
roads511: { enabled: boolean; tick_seconds: number; api_key: string; base_url: string; endpoints: string[]; bbox: number[]; feed_source?: FeedSource }
|
||||
firms: { enabled: boolean; tick_seconds: number; map_key: string; source: string; bbox: number[]; day_range: number; confidence_min: string; proximity_km: number; feed_source?: FeedSource }
|
||||
central?: { enabled: boolean; url: string; durable: string }
|
||||
central?: { enabled: boolean; url: string; durable: string; region: string }
|
||||
}
|
||||
|
||||
type FeedHealth = EnvStatus['feeds'][number]
|
||||
|
|
@ -406,6 +406,10 @@ export default function Environment() {
|
|||
<TextInput label="Durable" value={env.central.durable || ''}
|
||||
onChange={(v) => up({ central: { ...env.central!, durable: v } })}
|
||||
placeholder="meshai-v04" />
|
||||
<TextInput label="Region" value={env.central.region || ''}
|
||||
onChange={(v) => up({ central: { ...env.central!, region: v } })}
|
||||
placeholder="us.id"
|
||||
helper="Central v0.9.20 region token (dotted, e.g. 'us.id'). Empty = bare wildcards (all-US firehose)." />
|
||||
</div>
|
||||
</div>
|
||||
)}
|
||||
|
|
|
|||
|
|
@ -33,10 +33,11 @@ def consumer_config():
|
|||
return ConsumerConfig(deliver_policy=DeliverPolicy.NEW)
|
||||
|
||||
|
||||
# meshai adapter (env-config attr) -> Central subject filters it consumes.
|
||||
# Adapters with no Central equivalent (avalanche, ducting, roads511) are absent;
|
||||
# setting source=central on those subscribes to nothing (logged).
|
||||
ADAPTER_SUBJECTS: dict[str, list[str]] = {
|
||||
# Bare-wildcard subjects, pre-v0.9.20. Still used when `central.region` is
|
||||
# empty (backward-compat fallback) and as the canonical adapter -> family map.
|
||||
# Adapters with no Central equivalent (avalanche, ducting) are absent; flipping
|
||||
# those to source=central subscribes to nothing (logged).
|
||||
_SUBJECTS_BARE: dict[str, list[str]] = {
|
||||
"nws": ["central.wx.>"],
|
||||
"fires": ["central.fire.incident.>", "central.fire.perimeter.>"],
|
||||
"firms": ["central.fire.>"],
|
||||
|
|
@ -44,9 +45,59 @@ ADAPTER_SUBJECTS: dict[str, list[str]] = {
|
|||
"usgs": ["central.hydro.>"],
|
||||
"swpc": ["central.space.>"],
|
||||
"traffic": ["central.traffic.>"],
|
||||
"roads511": ["central.traffic.>"],
|
||||
"roads511": ["central.traffic.>"], # shared with traffic; sub-adapter routing
|
||||
}
|
||||
|
||||
# Backwards-compat: keep ADAPTER_SUBJECTS importable for legacy readers/tests.
|
||||
ADAPTER_SUBJECTS = _SUBJECTS_BARE
|
||||
|
||||
|
||||
def _subjects_for(adapter: str, region: Optional[str]) -> list[str]:
|
||||
"""Build region-aware Central subject filters for an adapter (v0.5.4).
|
||||
|
||||
Central v0.9.20 (shipped 2026-05-28) added per-region subject suffixes so
|
||||
consumers interested in a single region can have the firehose filtered
|
||||
server-side instead of dragging all-US events and discarding 95% locally.
|
||||
|
||||
`region` is a dotted token tree, e.g. 'us.id' for Idaho. Adapters use
|
||||
one of three suffix patterns; the v0.9.20 scheme is not uniform:
|
||||
|
||||
- region BEFORE the wildcard (nws):
|
||||
central.wx.alert.us.id.>
|
||||
- region AFTER the wildcard (quake / firms / usgs / traffic):
|
||||
central.quake.event.>.us.id
|
||||
central.fire.hotspot.>.us.id
|
||||
central.hydro.>.us.id (+ ".unknown" workaround, see below)
|
||||
central.traffic.>.id (state only, no us. prefix)
|
||||
- state-only token at a fixed depth (fires):
|
||||
central.fire.incident.<state>.>
|
||||
central.fire.perimeter.<state>.>
|
||||
- region ignored (swpc) — space weather is planetary.
|
||||
|
||||
The .unknown workaround: v0.9.20 leaves USGS hydro events whose gauge
|
||||
state can't be inferred on `central.hydro.>.unknown`. Subscribing to
|
||||
both avoids losing those rows until v0.9.20.1 backfills the state tag.
|
||||
|
||||
Empty/None region returns the bare-wildcard form (v0.5.3 behaviour).
|
||||
Adapters without a Central equivalent (avalanche, ducting) return [].
|
||||
"""
|
||||
if not region:
|
||||
return list(_SUBJECTS_BARE.get(adapter, []))
|
||||
state = region.split(".")[-1]
|
||||
table: dict[str, list[str]] = {
|
||||
"nws": [f"central.wx.alert.{region}.>"],
|
||||
"fires": [f"central.fire.incident.{state}.>",
|
||||
f"central.fire.perimeter.{state}.>"],
|
||||
"firms": [f"central.fire.hotspot.>.{region}"],
|
||||
"usgs_quake": [f"central.quake.event.>.{region}"],
|
||||
"usgs": [f"central.hydro.>.{region}",
|
||||
"central.hydro.>.unknown"],
|
||||
"swpc": ["central.space.>"],
|
||||
"traffic": [f"central.traffic.>.{state}"],
|
||||
"roads511": [f"central.traffic.>.{state}"], # shared with traffic
|
||||
}
|
||||
return list(table.get(adapter, []))
|
||||
|
||||
# Bridge between Central's adapter taxonomy and meshai's family-tab source names.
|
||||
# Central names some adapters differently (e.g. "wfigs_incidents" vs meshai's
|
||||
# "fires"); remap so dashboard per-adapter event filtering (which keys on the
|
||||
|
|
@ -164,16 +215,24 @@ class CentralConsumer:
|
|||
self._subs: list = []
|
||||
|
||||
# ---- subject derivation ----
|
||||
def _region(self) -> str:
|
||||
"""Active Central region (v0.5.4). Empty string = pre-v0.9.20 bare wildcards."""
|
||||
if self._central is None:
|
||||
return ""
|
||||
return getattr(self._central, "region", "") or ""
|
||||
|
||||
def _subject_owned(self) -> dict:
|
||||
"""Map each Central subject filter -> set of meshai source names (adapter
|
||||
attrs) that are feed_source=central and consume it. A shared subject
|
||||
(central.traffic.> for both traffic and roads511) carries multiple owned
|
||||
sources; _handle drops events whose remapped source isn't in the set."""
|
||||
(central.traffic.>.id for both traffic and roads511) carries multiple
|
||||
owned sources; _handle drops events whose remapped source isn't in the
|
||||
set. v0.5.4: subject shapes are region-aware via _subjects_for()."""
|
||||
region = self._region()
|
||||
owned: dict = {}
|
||||
for attr, subjects in ADAPTER_SUBJECTS.items():
|
||||
for attr in _SUBJECTS_BARE.keys():
|
||||
cfg = getattr(self._env, attr, None)
|
||||
if cfg is not None and getattr(cfg, "feed_source", "native") == "central":
|
||||
for subj in subjects:
|
||||
for subj in _subjects_for(attr, region):
|
||||
owned.setdefault(subj, set()).add(attr)
|
||||
for attr in ("avalanche", "ducting"):
|
||||
cfg = getattr(self._env, attr, None)
|
||||
|
|
@ -304,6 +363,9 @@ class CentralConsumer:
|
|||
sorted(subject_owned))
|
||||
return
|
||||
|
||||
region = self._region()
|
||||
logger.info("CentralConsumer: connecting region=%r subjects=%s",
|
||||
region or "(bare wildcards)", sorted(subject_owned))
|
||||
import nats # lazy: no NATS dependency at boot unless actually consuming
|
||||
self._nc = await nats.connect(
|
||||
self._central.url,
|
||||
|
|
|
|||
|
|
@ -440,12 +440,20 @@ class FIRMSConfig(_SourcedFeed):
|
|||
|
||||
@dataclass
|
||||
class CentralConsumerConfig:
|
||||
"""Connection settings for the Central NATS JetStream consumer (v0.4)."""
|
||||
"""Connection settings for the Central NATS JetStream consumer (v0.4).
|
||||
|
||||
v0.5.4 adds `region` — a dotted v0.9.20 region token (e.g. 'us.id' for
|
||||
Idaho) appended to each subscribed Central subject so the firehose is
|
||||
filtered server-side. Empty string falls back to bare wildcards (pre-
|
||||
v0.9.20 behaviour). One region applies to all central adapters; per-
|
||||
adapter overrides can land in v0.6.
|
||||
"""
|
||||
|
||||
enabled: bool = False
|
||||
url: str = "nats://central.echo6.mesh:4222"
|
||||
durable: str = "meshai-consumer"
|
||||
connect_timeout: float = 10.0
|
||||
region: str = "us.id"
|
||||
|
||||
|
||||
@dataclass
|
||||
|
|
|
|||
File diff suppressed because one or more lines are too long
|
|
@ -8,7 +8,7 @@
|
|||
<link rel="preconnect" href="https://fonts.googleapis.com">
|
||||
<link rel="preconnect" href="https://fonts.gstatic.com" crossorigin>
|
||||
<link href="https://fonts.googleapis.com/css2?family=JetBrains+Mono:wght@400;500;600;700&display=swap" rel="stylesheet">
|
||||
<script type="module" crossorigin src="/assets/index-DCFmSeOM.js"></script>
|
||||
<script type="module" crossorigin src="/assets/index-B24tHcYj.js"></script>
|
||||
<link rel="stylesheet" crossorigin href="/assets/index-DjhQa8Mv.css">
|
||||
</head>
|
||||
<body>
|
||||
|
|
|
|||
|
|
@ -54,7 +54,10 @@ def test_no_subjects_when_all_native():
|
|||
|
||||
|
||||
def test_subjects_when_central():
|
||||
# v0.5.4: assert the legacy bare-wildcard form by clearing region.
|
||||
# Region-aware subject shapes are covered by test_central_region_routing.py.
|
||||
c, env, rec = make_consumer()
|
||||
env.central.region = ""
|
||||
env.usgs_quake.feed_source = "central"
|
||||
assert "central.quake.>" in c.subjects()
|
||||
|
||||
|
|
|
|||
88
tests/test_central_region_routing.py
Normal file
88
tests/test_central_region_routing.py
Normal file
|
|
@ -0,0 +1,88 @@
|
|||
"""v0.5.4: Central v0.9.20 region-aware subject building.
|
||||
|
||||
Exercises `_subjects_for(adapter, region)` and the wiring through
|
||||
`CentralConsumer._subject_owned()`. The spec is hard-coded in the test
|
||||
strings on purpose so a future drift in the v0.9.20 subject scheme
|
||||
fails noisily here instead of silently shipping wrong filters.
|
||||
"""
|
||||
|
||||
from meshai.central.consumer import (
|
||||
CentralConsumer,
|
||||
_subjects_for,
|
||||
_SUBJECTS_BARE,
|
||||
)
|
||||
from meshai.config import EnvironmentalConfig
|
||||
|
||||
|
||||
# --------------------------------------------------------------------- per-adapter
|
||||
|
||||
def test_subjects_for_nws_us_id():
|
||||
"""NWS: region BEFORE wildcard (matches alert.<region>.<...>)."""
|
||||
assert _subjects_for("nws", "us.id") == ["central.wx.alert.us.id.>"]
|
||||
|
||||
|
||||
def test_subjects_for_usgs_quake_us_id():
|
||||
"""USGS quake: region AFTER wildcard."""
|
||||
assert _subjects_for("usgs_quake", "us.id") == ["central.quake.event.>.us.id"]
|
||||
|
||||
|
||||
def test_subjects_for_firms_us_id():
|
||||
"""FIRMS hotspots: region AFTER wildcard, hotspot domain explicit."""
|
||||
assert _subjects_for("firms", "us.id") == ["central.fire.hotspot.>.us.id"]
|
||||
|
||||
|
||||
def test_subjects_for_fires_us_id_uses_state_token():
|
||||
"""NIFC fires: state-only token at depth-4 for both incident + perimeter."""
|
||||
assert _subjects_for("fires", "us.id") == [
|
||||
"central.fire.incident.id.>",
|
||||
"central.fire.perimeter.id.>",
|
||||
]
|
||||
|
||||
|
||||
def test_subjects_for_traffic_and_roads511_share_state_token():
|
||||
"""Traffic family: bare-state suffix (no us. prefix), shared by both adapters."""
|
||||
assert _subjects_for("traffic", "us.id") == ["central.traffic.>.id"]
|
||||
assert _subjects_for("roads511", "us.id") == ["central.traffic.>.id"]
|
||||
|
||||
|
||||
def test_subjects_for_usgs_includes_unknown_workaround():
|
||||
"""USGS hydro: subscribes to BOTH the region-tagged filter and the
|
||||
".unknown" filter to cover gauges whose state Central v0.9.20 can't
|
||||
infer yet (workaround until v0.9.20.1 backfills the tag)."""
|
||||
assert _subjects_for("usgs", "us.id") == [
|
||||
"central.hydro.>.us.id",
|
||||
"central.hydro.>.unknown",
|
||||
]
|
||||
|
||||
|
||||
def test_subjects_for_swpc_stays_global():
|
||||
"""SWPC: space weather is planetary; region argument is ignored."""
|
||||
assert _subjects_for("swpc", "us.id") == ["central.space.>"]
|
||||
assert _subjects_for("swpc", "us.mt") == ["central.space.>"] # same regardless
|
||||
assert _subjects_for("swpc", "") == ["central.space.>"]
|
||||
|
||||
|
||||
# --------------------------------------------------------------------- backward compat
|
||||
|
||||
def test_subjects_for_empty_region_falls_back_to_bare_wildcards():
|
||||
"""Empty/None region = pre-v0.9.20 behaviour for every adapter, byte-identical
|
||||
to the legacy _SUBJECTS_BARE map. Adapters absent from the map return []."""
|
||||
for adapter, expected in _SUBJECTS_BARE.items():
|
||||
assert _subjects_for(adapter, "") == expected, f"empty region mismatch for {adapter}"
|
||||
assert _subjects_for(adapter, None) == expected, f"None region mismatch for {adapter}"
|
||||
# Unknown adapters return empty regardless of region.
|
||||
assert _subjects_for("ducting", "us.id") == []
|
||||
assert _subjects_for("avalanche", "") == []
|
||||
|
||||
|
||||
# --------------------------------------------------------------------- integration
|
||||
|
||||
def test_central_region_default_propagates_to_consumer_subjects():
|
||||
"""Default region = 'us.id': flipping nws to central → consumer subscribes
|
||||
to the region-aware subject, not the bare wildcard."""
|
||||
env = EnvironmentalConfig()
|
||||
assert env.central.region == "us.id" # spec default
|
||||
env.nws.feed_source = "central"
|
||||
so = CentralConsumer(env, None)._subject_owned()
|
||||
assert list(so.keys()) == ["central.wx.alert.us.id.>"]
|
||||
assert so["central.wx.alert.us.id.>"] == {"nws"}
|
||||
|
|
@ -17,8 +17,15 @@ def _envelope(adapter, category="x.y", eid="e1"):
|
|||
|
||||
def _route(central, adapter, subject, category="x.y"):
|
||||
"""Simulate a message arriving on the subscription that matches `subject`,
|
||||
with that subscription's owned-sources, and return the emitted Event (or None)."""
|
||||
with that subscription's owned-sources, and return the emitted Event (or None).
|
||||
|
||||
v0.5.4: this helper deliberately clears central.region so sub-adapter
|
||||
routing is exercised against bare wildcards (its concern is the
|
||||
owned-sources filter, not the region-aware subject shape — those are
|
||||
tested in test_central_region_routing.py).
|
||||
"""
|
||||
env = EnvironmentalConfig()
|
||||
env.central.region = ""
|
||||
for a in central:
|
||||
getattr(env, a).feed_source = "central"
|
||||
rec = []
|
||||
|
|
@ -71,7 +78,11 @@ def test_tomtom_incidents_remaps_to_traffic():
|
|||
|
||||
|
||||
def test_subject_owned_shares_traffic_subject():
|
||||
# v0.5.4: assert the legacy bare-wildcard shape by clearing region.
|
||||
# Region-aware shared-subject behaviour ('central.traffic.>.id' for both
|
||||
# traffic and roads511) is covered in test_central_region_routing.py.
|
||||
env = EnvironmentalConfig()
|
||||
env.central.region = ""
|
||||
env.traffic.feed_source = "central"
|
||||
env.roads511.feed_source = "central"
|
||||
so = CentralConsumer(env, None)._subject_owned()
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue