Eliminates the duplication that has been hand-bumped through PRs B, C, D, E.
Adding a stream is now one StreamEntry in src/central/streams.py + one
migration row in config.streams. supervisor STREAM_SUBJECTS / archive
STREAMS / gui DASHBOARD_STREAMS all derive at import time. No drift
possible because there is one source.
Pure refactor; no behavior change. Runtime verified: derived structures
are byte-equivalent to the previous literal definitions.
src/central/streams.py (new):
@dataclass(frozen=True)
class StreamEntry:
name: str
subject_filter: str
event_bearing: bool = True # archive consumes from this stream
dashboard: bool = True # GUI dashboard surfaces this stream
STREAMS: list[StreamEntry] = [
StreamEntry("CENTRAL_WX", "central.wx.>"),
StreamEntry("CENTRAL_FIRE", "central.fire.>"),
StreamEntry("CENTRAL_QUAKE", "central.quake.>"),
StreamEntry("CENTRAL_SPACE", "central.space.>"),
StreamEntry("CENTRAL_DISASTER", "central.disaster.>"),
StreamEntry("CENTRAL_META", "central.meta.>", event_bearing=False),
]
Consumers derive:
supervisor.STREAM_SUBJECTS = {s.name: [s.subject_filter] for s in STREAMS}
(includes META: supervisor must create every stream in JetStream)
archive.STREAMS = [(s.name, s.subject_filter) for s in STREAMS if s.event_bearing]
(excludes META: status messages, not events)
gui.DASHBOARD_STREAMS = [s.name for s in STREAMS if s.dashboard]
To resolve the name collision between the registry STREAMS and the
existing archive.STREAMS public symbol, archive.py imports the registry
under an alias: from central.streams import STREAMS as STREAM_REGISTRY.
The archives STREAMS surface (the tuple-list) is unchanged for callers.
Same alias used in supervisor.py and gui/routes.py for symmetry.
Migration files unchanged. config.streams keeps seeding retention/bytes --
operator-tunable ops state, separate SoT from the structural mapping.
Tests:
Dropped from test_archive_multi_stream.py (7, all tautological vs. registry):
test_streams_list_has_five_entries (magic-number count)
test_streams_contains_central_wx / fire / quake / space / disaster
test_streams_excludes_central_meta
Dropped from test_dashboard.py:
`assert len(streams) == 6` line inside test_single_stream_failure_doesnt_crash_card
(the test itself stays; only the magic-number assertion is removed)
Added in test_stream_registry.py (8 invariant tests):
test_stream_names_unique
test_subject_filters_unique
test_subject_filter_central_prefix_wildcard
test_meta_is_only_non_event_bearing
test_supervisor_stream_subjects_includes_meta
test_supervisor_stream_subjects_includes_all
test_archive_streams_excludes_non_event_bearing
test_dashboard_streams_matches_dashboard_flag
The new tests assert properties (uniqueness, format, derivation correctness),
not literals. Future stream additions need zero new test code -- every
invariant automatically covers them.
Note: test file named tests/test_stream_registry.py (not test_streams.py)
to avoid colliding with the pre-existing tests/test_streams.py, which
covers the GUI streams-management page.
Full suite: 427 passed (was 426 on main: -7 dropped + 8 added).
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Adds the GDACS (Global Disaster Alert and Coordination System) adapter
against the self-describing framework. Polls https://www.gdacs.org/xml/rss.xml
every 600s, parses the RSS items, and publishes to a new CENTRAL_DISASTER
JetStream stream on central.disaster.<eventtype_lower>.<country_lower>.
Locked decisions:
- Keep: WF, DR, FL, VO, TC. Drop: EQ (USGS canonical on central.quake.>),
plus any future-unknown eventtype.
- Filter via settings_schema event_types: list[str] so operators can
re-allow without a code change.
- Dedup by RSS guid (format <eventtype><eventid>, stable across reissue).
- Severity from gdacs:alertlevel (Green=1, Orange=2, Red=3, default 0).
- Fall-off uses GDACS gdacs:iscurrent=false as explicit tombstone signal,
with a fallback for items that vanish entirely from the feed. Tombstones
publish on disaster.removed.<eventtype>.<country>.
- Geo: centroid from geo:Point, bbox from gdacs:bbox (reordered to Geo
(minLon, minLat, maxLon, maxLat)), primary_region from gdacs:iso3.
CENTRAL_DISASTER stream: 7d retention, 1 GiB max_bytes, mirroring
CENTRAL_FIRE / CENTRAL_QUAKE / CENTRAL_SPACE. Migrations 020 (adapter row,
enabled=false, default event_types in settings) and 021 (stream seed).
STREAM_SUBJECTS, archive STREAMS, GUI DASHBOARD_STREAMS each pick up
the new stream.
Tests: 14 new in tests/test_gdacs.py using frozen RSS fixtures with WF/DR/EQ/XX
items (covering normalization, EQ drop, unknown drop, settings override,
guid dedup, iscurrent=false tombstone, missing-from-feed tombstone,
helper boundaries). Stream-count assertions bumped 4->5 and 5->6 for
the new stream (anti-pattern noted; queued as a follow-up PR E.5).
+1 membership test test_streams_contains_central_disaster.
Full suite: 426 passed.
End-to-end on CT104: 48 events published on first poll (44 disaster.wf +
4 disaster.fl), zero EQ events, all subjects under central.disaster.>
with lowercase-hyphenated country suffixes.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Three independent adapters sharing src/central/adapters/swpc_common.py,
mirroring the WFIGS two-adapter pattern. Each adapter has its own row in
config.adapters (ships disabled), its own cadence, and its own dedup
state, so operators can independently enable/disable and so a broken
upstream endpoint does not silently mask a healthy one.
Subjects:
swpc_alerts -> central.space.alert.<product_id_lower>
swpc_kindex -> central.space.kindex
swpc_protons -> central.space.proton_flux
Dedup keys:
alerts: product_id + issue_datetime
kindex: time_tag
protons: time_tag + energy
Severity: G-scale on product_id for K0[5-9][AW] alerts (G1-G5 -> 1-4),
G-scale on Kp for kindex, 0 for protons (raw flux carried in event.data).
No geo on any SWPC events (centroid=None, regions=[], primary_region=None).
No fall-off detection for alerts -- a single 115-row sample cannot confirm
whether alerts disappear from the upstream JSON when expired; deferred to
a later pass after 24h of observation.
CENTRAL_SPACE stream seeded with 7-day retention / 1 GiB max_bytes, mirroring
CENTRAL_FIRE / CENTRAL_QUAKE. STREAM_SUBJECTS, archive STREAMS, and
DASHBOARD_STREAMS each pick up the new stream.
Tests: 16 new cases in tests/test_swpc.py using real-shape frozen JSON
fixtures (alerts product_ids EF3A/K05A/K07A; kindex Kp boundaries; protons
composite dedup). Two existing tests updated for the new stream count
(test_archive_multi_stream.test_streams_list_has_three_entries renamed to
_has_four_entries; test_dashboard expects 5 streams not 4); added a
test_streams_contains_central_space companion.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Fix A - /dashboard/polls:
- Use get_last_msg instead of pull_subscribe (no durable consumers)
- Fix subject filter: central.meta.adapter.{name}.status
- Parse correct fields: ts and ok from status message
- Handle NotFoundError gracefully when no status exists
Fix B - CSRF exception handler:
- Add global CsrfProtectError handler in __init__.py
- Return friendly "session expired" message instead of 500
- Re-render forms with error or redirect to /login
- Update templates to display error messages
Tests:
- Add get_last_msg mocking tests for polls
- Add regression test verifying no pull_subscribe
- Add CSRF handler tests
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
