mirror of
https://github.com/zvx-echo6/meshai.git
synced 2026-08-26 17:31:34 +00:00
feat(adapters): WZDx daily work-zone summary + FIRMS restart-safe cold-start (#108)
1. WZDx work zones: replace per-event broadcasting with a once-a-day per-region count summary. Coalesce the upstream per-direction / per-schedule-day fan-out into one row per physical zone (road + lat3 + lon3 + sub_type); work zones are stored in traffic_events but no longer per-event broadcast, while 511 crash / closure / hazard incidents still broadcast live. A WZDxSummaryScheduler emits one count line per coverage region once a day (default 07:00 America/Boise), routed via the region_routes 'roads' cells; work-zone details are DM-queryable (build_work_zones_detail). New config: wzdx.summary_enabled / summary_time / summary_tz. 2. FIRMS cold-start is now restart-safe: gate the silent-seed on the persisted firms_pixels baseline being empty (first-ever run) instead of an in-memory per-boot flag, so a restart no longer silently absorbs a genuinely-new hotspot cluster. Co-authored-by: Matt Johnson <mj@k7zvx.com> Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
GitHub
parent
b0b0697bac
commit
15ddedf22b
13 changed files with 1503 additions and 14 deletions
|
|
@ -212,7 +212,8 @@ REGISTRY: dict[tuple[str, str], dict[str, Any]] = {
|
|||
"description": "Which sub_types to broadcast. Empty list = all.",
|
||||
},
|
||||
# =================================================================
|
||||
# WZDX -- 3 settings (broadcast gate, severity gate, sub-type filter)
|
||||
# WZDX -- 6 settings (broadcast gate, severity gate, sub-type filter,
|
||||
# per-region daily count summary enable/time/tz)
|
||||
# =================================================================
|
||||
("wzdx", "broadcast"): {
|
||||
"default": False,
|
||||
|
|
@ -229,6 +230,21 @@ REGISTRY: dict[tuple[str, str], dict[str, Any]] = {
|
|||
"type": "json",
|
||||
"description": "Work zone sub-types to broadcast. Empty = all.",
|
||||
},
|
||||
("wzdx", "summary_enabled"): {
|
||||
"default": True,
|
||||
"type": "bool",
|
||||
"description": "Enable the once-a-day per-region active work-zone count summary broadcast.",
|
||||
},
|
||||
("wzdx", "summary_time"): {
|
||||
"default": "07:00",
|
||||
"type": "str",
|
||||
"description": "Local HH:MM time the daily work-zone count summary fires.",
|
||||
},
|
||||
("wzdx", "summary_tz"): {
|
||||
"default": "America/Boise",
|
||||
"type": "str",
|
||||
"description": "Timezone the daily work-zone count summary's summary_time is interpreted in.",
|
||||
},
|
||||
|
||||
# =================================================================
|
||||
# CENTRAL consumer -- 1 setting (severity-int bucket boundaries)
|
||||
|
|
|
|||
42
work/meshai/env/firms.py
vendored
42
work/meshai/env/firms.py
vendored
|
|
@ -381,6 +381,35 @@ class FIRMSAdapter:
|
|||
return "MODIS"
|
||||
return self._source or "?"
|
||||
|
||||
def _firms_pixels_empty(self) -> bool:
|
||||
"""Is the PERSISTED ``firms_pixels`` baseline empty (first-ever run)?
|
||||
|
||||
Uses the SAME DB handle the fusion engine reaches ``firms_pixels``
|
||||
through (``persistence.get_db`` -- exactly what
|
||||
``firms_handler.ingest_hotspot_pixel`` opens). Returns True only when the
|
||||
table holds ZERO rows, which is the sole condition under which the
|
||||
cold-start silent-seed should fire.
|
||||
|
||||
Fail-safe on a truly-fresh DB where the table may not exist yet: a
|
||||
missing table (or any query error) is treated as empty -> True, so a
|
||||
genuine first-ever run still seeds silently and we NEVER crash the fetch.
|
||||
A NON-empty table (a restart with a real baseline) returns False, which
|
||||
collapses ``cold_start`` and lets genuinely-new clusters broadcast.
|
||||
"""
|
||||
try:
|
||||
from meshai.persistence import get_db
|
||||
conn = get_db()
|
||||
# EXISTS(...) short-circuits at the first row -> cheap even on a
|
||||
# large baseline; 0 -> empty, 1 -> at least one persisted pixel.
|
||||
row = conn.execute(
|
||||
"SELECT EXISTS(SELECT 1 FROM firms_pixels LIMIT 1)"
|
||||
).fetchone()
|
||||
return not (row and row[0])
|
||||
except Exception:
|
||||
# Missing table / persistence unavailable -> treat as empty so the
|
||||
# first-ever run seeds silently (and the fetch never crashes).
|
||||
return True
|
||||
|
||||
def _run_fusion(self, raw_events: list) -> list:
|
||||
"""Feed each fetched hotspot pixel into the SHARED attribution/fusion
|
||||
engine and collect the fire-fusion broadcasts.
|
||||
|
|
@ -405,7 +434,18 @@ class FIRMSAdapter:
|
|||
# Cold-start silent-seed gate: captured once for the whole batch BEFORE
|
||||
# the flag is flipped, so every pixel in the first full-day sweep seeds
|
||||
# together (mirrors store._ingest_fires cold_start capture, F1).
|
||||
cold_start = not self._firms_seeded
|
||||
#
|
||||
# RESTART-SAFE: the in-memory ``_firms_seeded`` flag resets to False on
|
||||
# every process start, so on its own it would silent-seed the ENTIRE
|
||||
# first post-restart batch -- absorbing genuinely-new clusters. Gate it
|
||||
# additionally on the PERSISTED baseline (``firms_pixels``) being empty,
|
||||
# mirroring WFIGS's ``first_sight = row is None``: cold-start seeding only
|
||||
# applies on the FIRST-EVER run (no persisted pixels). Any restart with an
|
||||
# existing baseline -> cold_start False -> pixels ingest with seed=False;
|
||||
# pre-existing pixels are naturally suppressed (already attributed /
|
||||
# cluster_broadcast_at-stamped / INSERT-OR-IGNORE no-ops) so NO burst,
|
||||
# while a genuinely-new cluster still broadcasts.
|
||||
cold_start = (not self._firms_seeded) and self._firms_pixels_empty()
|
||||
fusion: list = []
|
||||
for evt in raw_events:
|
||||
props = evt.get("properties", {}) or {}
|
||||
|
|
|
|||
93
work/meshai/env/wzdx.py
vendored
93
work/meshai/env/wzdx.py
vendored
|
|
@ -30,8 +30,22 @@ the wire is identical to what the Central-side parser would have produced
|
|||
(road, direction, mile posts, folded sub_type, impact, ends_at). ``to_event``
|
||||
then converts ``ends_at`` → ``ends_at_epoch`` exactly as the Phase-2 bridge
|
||||
does (``calendar.timegm`` of the naive datetime) and augments with lat/lon +
|
||||
a stable ``external_id`` (WZDx ``data_source_id`` + feature id) so the
|
||||
incident gating decider dedups correctly.
|
||||
a stable ``external_id``.
|
||||
|
||||
Coalescing
|
||||
----------
|
||||
FHWA WZDx feeds frequently publish MULTIPLE road_event features for what is
|
||||
physically the SAME work zone (e.g. one feature per direction / per
|
||||
schedule-day fan / per segment). To avoid one traffic_events row (and one
|
||||
gate decision) per feature, ``external_id`` is a COALESCING key derived from
|
||||
the feature's road + rounded lat/lon + folded sub_type:
|
||||
``wzdx_{road}|{lat:.3f}|{lon:.3f}|{sub_type}`` — NOT the raw
|
||||
``data_source_id:feature_id`` pair. Multiple features that resolve to the
|
||||
same key are merged in ``_fetch_all`` (earliest start_at, latest end_at)
|
||||
into a single stored event before being handed to ``to_event()``, so exactly
|
||||
one ``traffic_events`` row per physical zone reaches the incident gating
|
||||
decider (``event_id`` == ``external_id`` for wzdx, so ``_seen``/decider/
|
||||
restart-seed all dedup on the same coalesced value).
|
||||
"""
|
||||
|
||||
import calendar
|
||||
|
|
@ -256,6 +270,11 @@ class WZDxAdapter:
|
|||
if not any_success:
|
||||
return False # all feeds down — keep the last known good set
|
||||
|
||||
# Coalesce features that resolve to the SAME physical work zone
|
||||
# (same road + rounded lat/lon + sub_type) BEFORE the bbox filter, so
|
||||
# exactly one merged event per key reaches everything downstream.
|
||||
new_events = self._coalesce_events(new_events)
|
||||
|
||||
# Optional bbox filter.
|
||||
if self._bbox and len(self._bbox) == 4:
|
||||
west, south, east, north = self._bbox
|
||||
|
|
@ -278,6 +297,54 @@ class WZDxAdapter:
|
|||
logger.info("WZDx work zones updated: %d active", len(new_events))
|
||||
return changed
|
||||
|
||||
@staticmethod
|
||||
def _coalesce_events(events: list) -> list:
|
||||
"""Merge stored-event dicts that share the same coalescing key.
|
||||
|
||||
Multiple WZDx features (e.g. per-direction / per-schedule-day fans)
|
||||
commonly describe the SAME physical work zone. ``event_id`` (==
|
||||
``external_id``) already encodes the coalescing key
|
||||
(``wzdx_{road}|{lat}|{lon}|{sub_type}``), so merging is a plain
|
||||
group-by on that key: keep the first-seen record's fields, but widen
|
||||
the window to the EARLIEST ``start_at`` and LATEST ``end_at`` across
|
||||
every merged feature (ignoring ``None`` on either side), and keep
|
||||
"priority" severity if ANY merged feature is a full closure.
|
||||
Preserves first-seen order of keys.
|
||||
"""
|
||||
merged: dict = {}
|
||||
order: list = []
|
||||
for evt in events:
|
||||
key = evt["event_id"]
|
||||
if key not in merged:
|
||||
merged[key] = dict(evt)
|
||||
order.append(key)
|
||||
continue
|
||||
cur = merged[key]
|
||||
# Earliest start_at (ignore None).
|
||||
starts = [v for v in (cur.get("start_at"), evt.get("start_at")) if v is not None]
|
||||
if starts:
|
||||
cur["start_at"] = min(starts)
|
||||
# Latest end_at (ignore None). If EITHER side is None, keep None
|
||||
# (an open-ended / unknown end wins — never silently invent one).
|
||||
if cur.get("end_at") is None or evt.get("end_at") is None:
|
||||
cur["end_at"] = None
|
||||
else:
|
||||
cur["end_at"] = max(cur["end_at"], evt["end_at"])
|
||||
# Priority severity wins if either merged feature is full_closure.
|
||||
if evt.get("severity") == "priority":
|
||||
cur["severity"] = "priority"
|
||||
cur_n = cur.get("normalized") or {}
|
||||
evt_n = evt.get("normalized") or {}
|
||||
if evt_n.get("impact") == "full_closure":
|
||||
cur_n = dict(cur_n)
|
||||
cur_n["impact"] = "full_closure"
|
||||
cur["normalized"] = cur_n
|
||||
# expires: keep the later of the two (matches the widened end_at
|
||||
# window intent — never expire a still-open merged zone early).
|
||||
if cur.get("expires") is not None and evt.get("expires") is not None:
|
||||
cur["expires"] = max(cur["expires"], evt["expires"])
|
||||
return [merged[k] for k in order]
|
||||
|
||||
@staticmethod
|
||||
def _iter_features(fc) -> list:
|
||||
"""Yield features from a WZDx v4 GeoJSON FeatureCollection."""
|
||||
|
|
@ -316,6 +383,16 @@ class WZDxAdapter:
|
|||
``central_normalizer._parse_wzdx_federal`` (which reads either the
|
||||
raw ``core_details.*`` nesting or a flattened envelope). Returns None
|
||||
for non-work-zone / malformed / id-less features (never raises).
|
||||
|
||||
``external_id`` (== ``event_id``) is a COALESCING key —
|
||||
``wzdx_{road}|{lat:.3f}|{lon:.3f}|{sub_type}`` — derived from the
|
||||
PARSED road/sub_type (``n``, from ``_parse_wzdx_federal``) and the
|
||||
rounded coordinates, NOT the raw ``data_source_id:feature_id`` pair.
|
||||
This lets multiple upstream features describing the same physical
|
||||
zone (e.g. per-direction / per-schedule-day fans) collapse to one
|
||||
``traffic_events`` row (merged in ``_fetch_all``/``_coalesce_events``).
|
||||
A feature with coords but no road/sub_type still gets a stable key
|
||||
(``wzdx_|<lat>|<lon>|``) — acceptable, still coalesces consistently.
|
||||
"""
|
||||
try:
|
||||
if not isinstance(feat, dict):
|
||||
|
|
@ -332,12 +409,13 @@ class WZDxAdapter:
|
|||
if event_type and event_type != "work-zone":
|
||||
return None
|
||||
|
||||
# Stable external id: data_source_id + feature id.
|
||||
# Stable identity check: data_source_id or feature id must exist
|
||||
# (still required so an id-less feature is never stored), even
|
||||
# though the coalescing key itself does not use these values.
|
||||
data_source_id = cd.get("data_source_id") or props.get("data_source_id")
|
||||
feat_id = feat.get("id") or cd.get("id") or props.get("id")
|
||||
if not feat_id and not data_source_id:
|
||||
return None # no stable identity → cannot dedup
|
||||
external_id = ":".join(str(p) for p in (data_source_id, feat_id) if p)
|
||||
|
||||
# Coordinates: prop lat/lon if present, else geometry centroid.
|
||||
lat = props.get("latitude")
|
||||
|
|
@ -354,13 +432,18 @@ class WZDxAdapter:
|
|||
geo = {"centroid": centroid} if centroid else {}
|
||||
n = _parse_wzdx_federal(inner_data, geo)
|
||||
|
||||
# Coalescing key: road + rounded lat/lon + folded sub_type.
|
||||
road = n.get("road") or ""
|
||||
sub_type = n.get("sub_type") or ""
|
||||
external_id = f"wzdx_{road}|{round(float(lat), 3)}|{round(float(lon), 3)}|{sub_type}"
|
||||
|
||||
# start/end epochs for the incident-gating keys.
|
||||
start_at = self._iso_to_epoch(props.get("start_date") or cd.get("start_date"))
|
||||
end_at = self._iso_to_epoch(props.get("end_date") or cd.get("end_date"))
|
||||
|
||||
return {
|
||||
"source": "wzdx",
|
||||
"event_id": f"wzdx_{external_id}",
|
||||
"event_id": external_id,
|
||||
"event_type": "Work Zone",
|
||||
"severity": "priority" if n.get("impact") == "full_closure" else "routine",
|
||||
"lat": float(lat),
|
||||
|
|
|
|||
|
|
@ -372,6 +372,77 @@ class EnvReporter:
|
|||
lines.append(f" - {road} {direction} ({county}): {sub}{impact}{delay}, seen {when}")
|
||||
return "\n".join(lines)[:_block_cap()]
|
||||
|
||||
def build_work_zones_detail(self, *, region: Optional[str] = None,
|
||||
limit: int = 10,
|
||||
now: Optional[int] = None) -> str:
|
||||
"""Active WZDx work zones (road, location, impact, end date).
|
||||
|
||||
build_traffic_detail() deliberately excludes these -- it only sources
|
||||
tomtom_incidents/itd_511/state_511_atis AND filters ``state = 'ID'``,
|
||||
but native wzdx rows carry ``source='wzdx'`` and ``state=NULL`` (see
|
||||
env/wzdx.py::to_event canonical_data), so they never surfaced on the
|
||||
DM path. This is the dedicated wzdx reader, region-scoped when a
|
||||
region name is passed (matches a configured coverage area's .name via
|
||||
event_region_names -- same derivation the wzdx_summary scheduler and
|
||||
Dispatcher.dispatch_scheduled_roads_broadcast use).
|
||||
"""
|
||||
if not self._adapter_included("wzdx"):
|
||||
return ""
|
||||
now = now if now is not None else int(time.time())
|
||||
try: conn = self._conn_factory()
|
||||
except Exception: return ""
|
||||
|
||||
try:
|
||||
rows = conn.execute(
|
||||
"SELECT road, direction, sub_type, impact, lat, lon, end_at "
|
||||
"FROM traffic_events "
|
||||
"WHERE source='wzdx' AND (end_at IS NULL OR end_at >= ?) "
|
||||
"ORDER BY end_at ASC LIMIT ?",
|
||||
(now, limit),
|
||||
).fetchall()
|
||||
except Exception:
|
||||
return ""
|
||||
if not rows:
|
||||
return ""
|
||||
|
||||
if region:
|
||||
# Region scoping needs the live coverage areas + a synthetic event
|
||||
# per row; guarded so a lookup failure degrades to unscoped output
|
||||
# rather than dropping the whole block.
|
||||
try:
|
||||
from meshai.coverage_area import areas_from_config, event_region_names
|
||||
from meshai.notifications.events import make_event
|
||||
from meshai.config import load_config as _load_config
|
||||
cfg = _load_config()
|
||||
areas = areas_from_config(getattr(cfg, "coverage", None))
|
||||
scoped = []
|
||||
for r in rows:
|
||||
lat, lon = r["lat"], r["lon"]
|
||||
if lat is None or lon is None:
|
||||
continue
|
||||
ev = make_event(source="wzdx", category="work_zone",
|
||||
lat=float(lat), lon=float(lon))
|
||||
if region in (event_region_names(ev, areas) or []):
|
||||
scoped.append(r)
|
||||
rows = scoped
|
||||
except Exception:
|
||||
logger.exception(
|
||||
"env_reporter: work_zones region scoping failed for %r; "
|
||||
"returning unscoped block", region)
|
||||
if not rows:
|
||||
return ""
|
||||
|
||||
header = (f"ACTIVE WORK ZONES ({region}):" if region
|
||||
else "ACTIVE WORK ZONES (WZDx):")
|
||||
lines = [header]
|
||||
for r in rows:
|
||||
road = r["road"] or "road?"
|
||||
direction = f" {r['direction']}" if r["direction"] else ""
|
||||
impact = r["impact"] or (r["sub_type"] or "work zone")
|
||||
ends = _fmt_epoch(r["end_at"]) if r["end_at"] else "no end date"
|
||||
lines.append(f" - {road}{direction}: {impact}, ends {ends}")
|
||||
return "\n".join(lines)[:_block_cap()]
|
||||
|
||||
def build_gauges_detail(self, *, limit: int = 10,
|
||||
now: Optional[int] = None) -> str:
|
||||
if not self._adapter_included("usgs_nwis"):
|
||||
|
|
@ -591,6 +662,7 @@ class EnvReporter:
|
|||
self.build_alerts_detail(now=now),
|
||||
self.build_quakes_detail(now=now),
|
||||
self.build_traffic_detail(now=now),
|
||||
self.build_work_zones_detail(now=now),
|
||||
self.build_gauges_detail(now=now),
|
||||
self.build_swpc_detail(now=now),
|
||||
self.build_satpass_detail(now=now),
|
||||
|
|
|
|||
|
|
@ -8,12 +8,22 @@ Mirrors incident_handler change-detection EXACTLY:
|
|||
* magnitude stepped up OR delay doubled OR icon changed → Update
|
||||
* otherwise → suppress
|
||||
|
||||
Work-zone STORE-not-BROADCAST rule (Part 2, wzdx daily-summary feature):
|
||||
A work zone (source=="wzdx" OR data["sub_type"]=="road_works") is ALWAYS
|
||||
persisted into traffic_events exactly like any other incident/roads row —
|
||||
the per-region daily count summary (notifications/scheduled/wzdx_summary.py)
|
||||
reads that table — but is NEVER broadcast per-event. 511 crash/closure/
|
||||
hazard rows (different sub_types) are completely unaffected and keep
|
||||
broadcasting exactly as before. last_broadcast_at is left NULL (never
|
||||
armed) for a work zone; there is no commit for it, so it can never look
|
||||
"already broadcast" for cold-start purposes on a different code path.
|
||||
|
||||
decide(data, *, source, now) -> GateResult:
|
||||
|
||||
Canonical data schema consumed (same keys as the central bridge
|
||||
produces; subset used by gating):
|
||||
external_id str | None — None for native adapters (always broadcast)
|
||||
source str — "tomtom_incidents" | "state_511_atis" | "itd_511" | …
|
||||
source str — "tomtom_incidents" | "state_511_atis" | "itd_511" | "wzdx" | …
|
||||
sub_type str | None
|
||||
road str | None
|
||||
direction str | None
|
||||
|
|
@ -41,8 +51,10 @@ decide(data, *, source, now) -> GateResult:
|
|||
Safe to call N times (all writes are conditional UPSERTs).
|
||||
|
||||
Native adapters (external_id is None or empty):
|
||||
No traffic_events lookup; always broadcast with lifecycle="native".
|
||||
The event bus inhibitor + group_key handle dedup for native events.
|
||||
No traffic_events lookup; always broadcast with lifecycle="native"
|
||||
(unless it's a work zone — see the work-zone rule above, which is
|
||||
checked here too for safety even though wzdx always carries an
|
||||
external_id in practice).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
|
|
@ -74,8 +86,20 @@ def decide(data: dict, *, source: str, now: float) -> GateResult:
|
|||
external_id = data.get("external_id") or None
|
||||
source_val = data.get("source") or source
|
||||
|
||||
# Work-zone discriminator (Part 2): native wzdx rows OR any itd_511 row
|
||||
# whose sub_type is the work-zone one. Computed once, used at every
|
||||
# broadcast-return point below (row is always written; only the
|
||||
# broadcast is suppressed for a work zone).
|
||||
_is_work_zone = (source_val == "wzdx") or (str(data.get("sub_type") or "") == "road_works")
|
||||
|
||||
# ── Native adapters: no external_id, no traffic_events dedup ─────────
|
||||
if not external_id:
|
||||
if _is_work_zone:
|
||||
return GateResult(
|
||||
broadcast=False,
|
||||
lifecycle="suppress",
|
||||
reason="work_zone stored, summary-only (native, no external_id)",
|
||||
)
|
||||
return GateResult(
|
||||
broadcast=True,
|
||||
lifecycle="native",
|
||||
|
|
@ -138,6 +162,14 @@ def decide(data: dict, *, source: str, now: float) -> GateResult:
|
|||
logger.exception("incident decide: INSERT failed for %s|%s",
|
||||
source_val, external_id)
|
||||
|
||||
# Work zone: row is persisted (above); never broadcast per-event.
|
||||
if _is_work_zone:
|
||||
return GateResult(
|
||||
broadcast=False,
|
||||
lifecycle="suppress",
|
||||
reason=f"work_zone stored, summary-only (new row {source_val}|{external_id})",
|
||||
)
|
||||
|
||||
patch = {"is_update": False, "_dedup_suffix": ""}
|
||||
|
||||
def _commit_new(committed_at: float) -> None:
|
||||
|
|
@ -173,6 +205,14 @@ def decide(data: dict, *, source: str, now: float) -> GateResult:
|
|||
except Exception:
|
||||
logger.exception("incident decide: UPDATE last_seen_at failed")
|
||||
|
||||
# Work zone: row refreshed (above); never broadcast per-event.
|
||||
if _is_work_zone:
|
||||
return GateResult(
|
||||
broadcast=False,
|
||||
lifecycle="suppress",
|
||||
reason=f"work_zone stored, summary-only (existing row {source_val}|{external_id})",
|
||||
)
|
||||
|
||||
last_bcast_at = row["last_broadcast_at"]
|
||||
|
||||
# ── Cold-start: row exists but was never delivered ────────────────────
|
||||
|
|
|
|||
|
|
@ -33,6 +33,12 @@ try:
|
|||
)
|
||||
except ImportError:
|
||||
BandConditionsScheduler = None
|
||||
try:
|
||||
from meshai.notifications.scheduled.wzdx_summary import (
|
||||
WZDxSummaryScheduler,
|
||||
)
|
||||
except ImportError:
|
||||
WZDxSummaryScheduler = None
|
||||
try:
|
||||
from meshai.notifications.reminders import ReminderScheduler
|
||||
except ImportError:
|
||||
|
|
@ -248,6 +254,25 @@ async def start_pipeline(bus: EventBus, config) -> DigestScheduler:
|
|||
_lg.getLogger("meshai.pipeline").exception(
|
||||
"band_conditions scheduler failed to start")
|
||||
|
||||
# Part 3: wzdx per-region daily work-zone count summary scheduler --
|
||||
# spawn alongside band_conditions. Best-effort: failures must NOT break
|
||||
# notifications pipeline startup. Respects adapter_config.wzdx.
|
||||
# summary_enabled at fire time (the scheduler itself checks it every
|
||||
# loop iteration); the try/except here only guards start()/construction.
|
||||
if WZDxSummaryScheduler is not None:
|
||||
try:
|
||||
comps = getattr(bus, "_pipeline_components", {}) or {}
|
||||
disp = comps.get("dispatcher")
|
||||
if disp is not None:
|
||||
wz_sched = WZDxSummaryScheduler(config, disp)
|
||||
await wz_sched.start()
|
||||
comps["wzdx_summary_scheduler"] = wz_sched
|
||||
bus._pipeline_components = comps
|
||||
except Exception:
|
||||
import logging as _lg
|
||||
_lg.getLogger("meshai.pipeline").exception(
|
||||
"wzdx_summary scheduler failed to start")
|
||||
|
||||
# v0.6-phase3 ReminderScheduler -- runs alongside band_conditions.
|
||||
if ReminderScheduler is not None:
|
||||
try:
|
||||
|
|
|
|||
|
|
@ -1121,6 +1121,187 @@ class Dispatcher:
|
|||
ch_type)
|
||||
return delivered_any
|
||||
|
||||
async def dispatch_scheduled_roads_broadcast(
|
||||
self, text: str, *,
|
||||
source_event_pk: str,
|
||||
lat=None, lon=None, region=None,
|
||||
) -> bool:
|
||||
"""Region-aware scheduled broadcast for the wzdx per-region DAILY work-
|
||||
zone count summary (Part 3).
|
||||
|
||||
Mirrors dispatch_scheduled_fire_broadcast byte-for-byte except it
|
||||
routes through the `roads` toggle + the region_routes matrix's
|
||||
"roads" family cells instead of "fire" / the `fire` toggle. One call
|
||||
per region-with-zones; the caller (WZDxSummaryScheduler) passes a
|
||||
representative (lat, lon) for that region so event_region_names
|
||||
re-derives the SAME region here that the caller already computed.
|
||||
|
||||
Cold-start grace still applies (consistent with the other scheduled
|
||||
broadcasts). Returns True on at least one successful mesh delivery.
|
||||
"""
|
||||
# Cold-start grace (mirrors dispatch_scheduled_fire_broadcast).
|
||||
grace_s = int(getattr(self._config.notifications,
|
||||
"cold_start_grace_seconds", 60) or 0)
|
||||
if grace_s > 0:
|
||||
now_anchor = time.time()
|
||||
if self._first_event_at is None:
|
||||
self._first_event_at = now_anchor
|
||||
self._persist_state()
|
||||
if (now_anchor - self._first_event_at) < grace_s:
|
||||
self._cold_start_dropped += 1
|
||||
self._persist_state()
|
||||
self._logger.info(
|
||||
"cold-start grace: dropping scheduled roads broadcast "
|
||||
"(pk=%s)", source_event_pk)
|
||||
return False
|
||||
|
||||
toggles = getattr(self._config.notifications, "toggles", None) or {}
|
||||
roads_tog = toggles.get("roads") if isinstance(toggles, dict) else None
|
||||
if roads_tog is None:
|
||||
self._logger.info(
|
||||
"scheduled-roads-broadcast: roads toggle not found; dropping "
|
||||
"(pk=%s)", source_event_pk)
|
||||
return False
|
||||
|
||||
# Build a synthetic roads Event purely to (a) run region derivation
|
||||
# and (b) reuse make_payload_from_event. category work_zone maps to
|
||||
# the `roads` family; severity priority mirrors the live fire path.
|
||||
from meshai.notifications.events import make_event, make_payload_from_event
|
||||
ev = make_event(
|
||||
source="wzdx", category="work_zone",
|
||||
severity="priority", title=text,
|
||||
lat=(float(lat) if lat is not None else None),
|
||||
lon=(float(lon) if lon is not None else None),
|
||||
)
|
||||
ev.data["_meshai_precomposed"] = True
|
||||
|
||||
# Derive regions from the representative location the SAME way
|
||||
# CoverageFilter does for a live event (named coverage areas ->
|
||||
# region names). Never raises: an unlocatable summary line simply
|
||||
# yields no regions -> toggle default.
|
||||
derived_regions: list = []
|
||||
try:
|
||||
from meshai.coverage_area import (
|
||||
areas_from_config, event_region_names,
|
||||
)
|
||||
_areas = areas_from_config(getattr(self._config, "coverage", None))
|
||||
if _areas and ev.lat is not None and ev.lon is not None:
|
||||
derived_regions = event_region_names(ev, _areas) or []
|
||||
except Exception:
|
||||
self._logger.exception(
|
||||
"scheduled-roads-broadcast: region derivation failed; "
|
||||
"falling back to toggle default (pk=%s)", source_event_pk)
|
||||
if derived_regions:
|
||||
ev.regions = list(derived_regions)
|
||||
ev.region = derived_regions[0]
|
||||
|
||||
# Resolve the channel plan. Each entry: (ch_type, chan_val).
|
||||
# ch_type in {"mesh_broadcast", "meshcore_broadcast"}
|
||||
# chan_val = Meshtastic channel index or MeshCore channel name.
|
||||
# _mt_owned / _mc_owned track per-transport matrix authority so an
|
||||
# owned transport does NOT also emit the toggle default.
|
||||
plan: list = []
|
||||
_mt_owned = False
|
||||
_mc_owned = False
|
||||
rr = getattr(self._config.notifications, "region_routes", None)
|
||||
_mt_on = bool(getattr(rr, "mt_enabled", False)) if rr is not None else False
|
||||
_mc_on = bool(getattr(rr, "mc_enabled", False)) if rr is not None else False
|
||||
if rr is not None and (_mt_on or _mc_on) and derived_regions:
|
||||
fam_cells = (getattr(rr, "cells", None) or {}).get("roads") or {}
|
||||
_seen: set = set()
|
||||
for _rg in derived_regions:
|
||||
if _rg in _seen or _rg not in fam_cells:
|
||||
continue
|
||||
_seen.add(_rg)
|
||||
_cell = fam_cells[_rg]
|
||||
# A matched region marks each ENABLED transport as matrix-owned
|
||||
# (mirrors the event path's authoritative-suppress), so it does
|
||||
# NOT fall back to the toggle default even if the column is null.
|
||||
if _mt_on:
|
||||
_mt_owned = True
|
||||
if _mc_on:
|
||||
_mc_owned = True
|
||||
_cell_enabled = (_cell.get("enabled", True) if isinstance(_cell, dict)
|
||||
else getattr(_cell, "enabled", True))
|
||||
if not _cell_enabled:
|
||||
continue
|
||||
_mt = (_cell.get("mt") if isinstance(_cell, dict)
|
||||
else getattr(_cell, "mt", None))
|
||||
_mc = (_cell.get("mc") if isinstance(_cell, dict)
|
||||
else getattr(_cell, "mc", None))
|
||||
if _mt_on and _mt is not None:
|
||||
plan.append(("mesh_broadcast", _mt))
|
||||
if _mc_on and _mc: # truthy: non-empty string
|
||||
plan.append(("meshcore_broadcast", _mc))
|
||||
|
||||
# Toggle-default fallback for any transport the matrix did NOT own.
|
||||
# Uses the roads toggle's priority severity_channels (band-conditions
|
||||
# is NOT consulted). A transport already owned by the matrix is skipped.
|
||||
sev_channels = getattr(roads_tog, "severity_channels", {}) or {}
|
||||
default_ch_types = [
|
||||
c for c in sev_channels.get("priority", ["mesh_broadcast"])
|
||||
if c in ("mesh_broadcast", "meshcore_broadcast")
|
||||
]
|
||||
for ct in default_ch_types:
|
||||
if ct == "mesh_broadcast" and not _mt_owned:
|
||||
plan.append(("mesh_broadcast",
|
||||
getattr(roads_tog, "broadcast_channel", None) or 0))
|
||||
elif ct == "meshcore_broadcast" and not _mc_owned:
|
||||
_mcn = getattr(roads_tog, "meshcore_channel", None)
|
||||
if _mcn:
|
||||
plan.append(("meshcore_broadcast", _mcn))
|
||||
|
||||
if not plan:
|
||||
self._logger.info(
|
||||
"scheduled-roads-broadcast: no channels resolved for pk=%s "
|
||||
"(regions=%s); dropping", source_event_pk, derived_regions)
|
||||
return False
|
||||
|
||||
delivered_any = False
|
||||
for ch_type, chan_val in plan:
|
||||
rule = self._toggle_to_rule(
|
||||
roads_tog, ch_type, ev,
|
||||
mt_override=(chan_val if ch_type == "mesh_broadcast" else None),
|
||||
mc_override=(chan_val if ch_type == "meshcore_broadcast" else None),
|
||||
)
|
||||
try:
|
||||
channel = self._channel_factory(rule, self._connector)
|
||||
payload = make_payload_from_event(ev, message=text)
|
||||
success = await channel.deliver(payload, rule)
|
||||
except Exception:
|
||||
self._logger.exception(
|
||||
"scheduled-roads-broadcast: delivery raised for %s", ch_type)
|
||||
success = False
|
||||
|
||||
if success:
|
||||
delivered_any = True
|
||||
self._logger.info(
|
||||
"scheduled-roads-broadcast: dispatched pk=%s via %s ch=%s "
|
||||
"regions=%s", source_event_pk, ch_type, chan_val,
|
||||
derived_regions or "DEFAULT")
|
||||
|
||||
# v20 per-mesh audit row (one per channel; mirrors
|
||||
# dispatch_scheduled_fire_broadcast).
|
||||
try:
|
||||
from meshai.persistence import get_db
|
||||
conn = get_db()
|
||||
bytes_sent = len(text.encode("utf-8")) if text else 0
|
||||
transport, channel_id, recipient = self._audit_route(rule, ch_type)
|
||||
conn.execute(
|
||||
"INSERT INTO mesh_broadcasts_out(sent_at, recipient, "
|
||||
"channel, text, source_event_table, source_event_pk, "
|
||||
"bytes_sent, ack_received, transport, success) "
|
||||
"VALUES (?,?,?,?,?,?,?,?,?,?)",
|
||||
(int(time.time()), recipient, channel_id, text,
|
||||
"traffic_events", str(source_event_pk), bytes_sent, 0,
|
||||
transport, 1 if success else 0),
|
||||
)
|
||||
except Exception:
|
||||
self._logger.exception(
|
||||
"scheduled-roads-broadcast: audit row insert failed for %s",
|
||||
ch_type)
|
||||
return delivered_any
|
||||
|
||||
@staticmethod
|
||||
def _audit_route(rule, ch_type: str):
|
||||
"""Resolve (transport, channel_id, recipient) for a mesh delivery.
|
||||
|
|
|
|||
229
work/meshai/notifications/scheduled/wzdx_summary.py
Normal file
229
work/meshai/notifications/scheduled/wzdx_summary.py
Normal file
|
|
@ -0,0 +1,229 @@
|
|||
"""wzdx per-region DAILY work-zone count summary scheduler (Part 3).
|
||||
|
||||
Fires ONCE PER DAY at a configurable local time (``adapter_config.wzdx.
|
||||
summary_time`` / ``summary_tz``, defaults "07:00" / "America/Boise"). For
|
||||
EACH coverage region with >=1 active in-coverage work zone, broadcasts ONE
|
||||
line:
|
||||
|
||||
"🚧 <Region Name>: <N> active work zones — DM AIDA for details"
|
||||
|
||||
Data source: the coalesced ``traffic_events(source='wzdx')`` rows written by
|
||||
the native WZDx adapter (env/wzdx.py) via the incident gating decider
|
||||
(notifications/gating/incident.py) — one row per PHYSICAL work zone (Part 1
|
||||
coalescing). A row counts toward EVERY coverage region its (lat, lon)
|
||||
resolves to (event_region_names over areas_from_config(config.coverage)); a
|
||||
region with zero matching rows is skipped (no line emitted).
|
||||
|
||||
Routing: each region's line is dispatched through
|
||||
``Dispatcher.dispatch_scheduled_roads_broadcast`` (Part 3's "roads"-family
|
||||
counterpart to ``dispatch_scheduled_fire_broadcast``), so it lands on the
|
||||
SAME channels a live work-zone event for that region would (region_routes
|
||||
matrix "roads" cells, falling back to the `roads` toggle default).
|
||||
|
||||
Deliberately NO change-detection / throttle / dedup table (per spec): a
|
||||
fixed daily slot is the whole mechanism. The clock-driven shape (next-slot
|
||||
loop, `fire_slot`) mirrors ``BandConditionsScheduler`` in band_conditions.py
|
||||
so operators reading one scheduler understand the other. A public
|
||||
``fire_once(now=None)`` lets tests drive a single pass deterministically
|
||||
without the sleep loop, analogous to ``fire_slot``/``tick_once`` on the
|
||||
other schedulers.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import time
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from typing import Any, Callable, Optional
|
||||
|
||||
try:
|
||||
from zoneinfo import ZoneInfo
|
||||
except ImportError:
|
||||
ZoneInfo = None # pragma: no cover (3.9+ only)
|
||||
|
||||
from meshai.adapter_config import adapter_config
|
||||
from meshai.notifications.scheduled.band_conditions import slot_epoch
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# ========================================================================
|
||||
# Public API — pure helpers (no DB / dispatcher), easy to unit test.
|
||||
# ========================================================================
|
||||
|
||||
|
||||
def format_wzdx_summary_line(region_name: str, count: int) -> str:
|
||||
"""One <=140-char broadcast line for a region's active work-zone count."""
|
||||
return f"🚧 {region_name}: {count} active work zones — DM AIDA for details"
|
||||
|
||||
|
||||
def count_active_work_zones_by_region(rows: list, areas: list) -> dict:
|
||||
"""Group coalesced wzdx rows by coverage region name -> count.
|
||||
|
||||
``rows`` are sqlite3.Row-like objects (or dicts) with at least
|
||||
``lat``/``lon`` keys. A row is counted in EVERY region its point
|
||||
resolves to (event_region_names over the named coverage areas) — a
|
||||
zone straddling two regions' bboxes counts in both. Rows that resolve
|
||||
to no named region are simply not counted anywhere (never raises,
|
||||
never invents a region).
|
||||
"""
|
||||
from meshai.coverage_area import event_region_names
|
||||
from meshai.notifications.events import make_event
|
||||
|
||||
counts: dict = {}
|
||||
# Representative (lat, lon) per region — the FIRST zone seen for that
|
||||
# region — so the caller can re-derive the same region deterministically
|
||||
# when dispatching (event_region_names(same point, areas) == region).
|
||||
rep_point: dict = {}
|
||||
for row in rows:
|
||||
try:
|
||||
lat = row["lat"] if not isinstance(row, dict) else row.get("lat")
|
||||
lon = row["lon"] if not isinstance(row, dict) else row.get("lon")
|
||||
except (IndexError, KeyError):
|
||||
lat = lon = None
|
||||
if lat is None or lon is None:
|
||||
continue
|
||||
ev = make_event(source="wzdx", category="work_zone", severity="routine",
|
||||
lat=float(lat), lon=float(lon))
|
||||
for region in event_region_names(ev, areas) or []:
|
||||
counts[region] = counts.get(region, 0) + 1
|
||||
rep_point.setdefault(region, (float(lat), float(lon)))
|
||||
return counts, rep_point
|
||||
|
||||
|
||||
# ========================================================================
|
||||
# Scheduler — async loop firing once per day
|
||||
# ========================================================================
|
||||
|
||||
|
||||
class WZDxSummaryScheduler:
|
||||
"""Fires the per-region wzdx daily work-zone count summary."""
|
||||
|
||||
def __init__(self, config, dispatcher, *,
|
||||
clock: Optional[Callable[[], float]] = None,
|
||||
sleep: Optional[Callable[[float], Any]] = None,
|
||||
tz_name: Optional[str] = None):
|
||||
self._config = config
|
||||
self._dispatcher = dispatcher
|
||||
self._clock = clock or time.time
|
||||
self._sleep = sleep or asyncio.sleep
|
||||
self._task: Optional[asyncio.Task] = None
|
||||
self._stop_event: Optional[asyncio.Event] = None
|
||||
self._tz_name = tz_name or str(adapter_config.wzdx.summary_tz)
|
||||
self._logger = logging.getLogger("meshai.scheduled.wzdx_summary")
|
||||
|
||||
def _enabled(self) -> bool:
|
||||
return bool(adapter_config.wzdx.summary_enabled)
|
||||
|
||||
def _summary_time(self) -> str:
|
||||
hh_mm = str(adapter_config.wzdx.summary_time or "07:00")
|
||||
return hh_mm if ":" in hh_mm else "07:00"
|
||||
|
||||
async def start(self) -> None:
|
||||
if self._task is not None and not self._task.done():
|
||||
raise RuntimeError("WZDxSummaryScheduler already running")
|
||||
self._stop_event = asyncio.Event()
|
||||
self._task = asyncio.create_task(self._run(),
|
||||
name="wzdx-summary-scheduler")
|
||||
self._logger.info(
|
||||
"WZDx summary scheduler started: enabled=%s time=%s tz=%s",
|
||||
self._enabled(), self._summary_time(), self._tz_name)
|
||||
|
||||
async def stop(self) -> None:
|
||||
if self._stop_event: self._stop_event.set()
|
||||
if self._task: await self._task
|
||||
|
||||
async def _run(self) -> None:
|
||||
while not (self._stop_event and self._stop_event.is_set()):
|
||||
if not self._enabled():
|
||||
await self._sleep(60); continue
|
||||
now = self._clock()
|
||||
now_dt = datetime.fromtimestamp(now, tz=timezone.utc)
|
||||
target_epoch = self._next_slot(now_dt)
|
||||
wait_s = max(1, target_epoch - int(now))
|
||||
try: await self._sleep(min(wait_s, 3600))
|
||||
except asyncio.CancelledError: break
|
||||
now2 = int(self._clock())
|
||||
if now2 >= target_epoch:
|
||||
await self.fire_slot(target_epoch)
|
||||
|
||||
def _next_slot(self, now_dt: datetime) -> int:
|
||||
"""Return the epoch of the next future daily slot."""
|
||||
hh_mm = self._summary_time()
|
||||
today_now = int(now_dt.timestamp())
|
||||
ep = slot_epoch(now_dt, hh_mm, self._tz_name)
|
||||
if ep > today_now:
|
||||
return ep
|
||||
tomorrow = now_dt + timedelta(days=1)
|
||||
return slot_epoch(tomorrow, hh_mm, self._tz_name)
|
||||
|
||||
async def fire_slot(self, slot_epoch_s: int) -> int:
|
||||
"""Compute + broadcast the daily summary for this slot. Returns the
|
||||
number of region lines dispatched."""
|
||||
return await self.fire_once(now=slot_epoch_s)
|
||||
|
||||
async def fire_once(self, now: Optional[float] = None) -> int:
|
||||
"""One pass: query coalesced active wzdx rows, group by region,
|
||||
dispatch one line per region-with-zones. Public so tests can drive
|
||||
it deterministically without the sleep loop (mirrors fire_slot/
|
||||
tick_once on the other schedulers). Returns the number of region
|
||||
lines dispatched (0 when there are no regions with >=1 active zone,
|
||||
or when coverage areas / rows are unavailable)."""
|
||||
now = now if now is not None else self._clock()
|
||||
now_int = int(now)
|
||||
|
||||
try:
|
||||
from meshai.persistence import get_db
|
||||
conn = get_db()
|
||||
except Exception:
|
||||
self._logger.warning("wzdx-summary: DB unavailable; skipping")
|
||||
return 0
|
||||
|
||||
try:
|
||||
rows = conn.execute(
|
||||
"SELECT road, lat, lon, sub_type, impact, end_at "
|
||||
"FROM traffic_events "
|
||||
"WHERE source='wzdx' AND (end_at IS NULL OR end_at >= ?)",
|
||||
(now_int,),
|
||||
).fetchall()
|
||||
except Exception:
|
||||
self._logger.exception("wzdx-summary: query failed; skipping")
|
||||
return 0
|
||||
|
||||
if not rows:
|
||||
return 0
|
||||
|
||||
try:
|
||||
from meshai.coverage_area import areas_from_config
|
||||
areas = areas_from_config(getattr(self._config, "coverage", None))
|
||||
except Exception:
|
||||
self._logger.exception("wzdx-summary: coverage areas load failed; skipping")
|
||||
return 0
|
||||
|
||||
if not areas:
|
||||
return 0
|
||||
|
||||
counts, rep_point = count_active_work_zones_by_region(rows, areas)
|
||||
if not counts:
|
||||
return 0
|
||||
|
||||
dispatched = 0
|
||||
for region, n in counts.items():
|
||||
if n < 1:
|
||||
continue
|
||||
line = format_wzdx_summary_line(region, n)
|
||||
rep_lat, rep_lon = rep_point.get(region, (None, None))
|
||||
pk = f"wzdx_summary_{region}_{now_int}"
|
||||
try:
|
||||
ok = await self._dispatcher.dispatch_scheduled_roads_broadcast(
|
||||
text=line,
|
||||
source_event_pk=pk,
|
||||
lat=rep_lat, lon=rep_lon, region=region,
|
||||
)
|
||||
except Exception:
|
||||
self._logger.exception(
|
||||
"wzdx-summary: dispatch raised for region=%s", region)
|
||||
ok = False
|
||||
if ok:
|
||||
dispatched += 1
|
||||
return dispatched
|
||||
|
|
@ -146,7 +146,12 @@ def test_tick_survives_socrata_url_object(adapter, monkeypatch):
|
|||
monkeypatch.setattr(adapter, "_http_get_json", fake_get)
|
||||
# Must not raise; the unwrapped feed URL is discovered and fetched.
|
||||
assert adapter.tick() is True
|
||||
assert {e["external_id"] for e in adapter.get_events()} == {"idot-1:A"}
|
||||
# Coalescing key (Part 1): wzdx_{road}|{lat:.3f}|{lon:.3f}|{sub_type} --
|
||||
# no longer the raw data_source_id:feat_id pair, so feat_id="A" doesn't
|
||||
# appear in the key.
|
||||
assert {e["external_id"] for e in adapter.get_events()} == {
|
||||
"wzdx_US-95|43.63|-116.915|lanes reduced, surface work"
|
||||
}
|
||||
|
||||
|
||||
# ============================================================
|
||||
|
|
@ -173,10 +178,13 @@ def test_canonical_data_core_fields(adapter):
|
|||
|
||||
|
||||
def test_canonical_external_id_is_stable(adapter):
|
||||
"""external_id combines data_source_id + feature id for gating dedup."""
|
||||
"""external_id is the Part-1 coalescing key: wzdx_{road}|{lat:.3f}|
|
||||
{lon:.3f}|{sub_type} -- NOT the raw data_source_id:feature_id pair, so
|
||||
multiple upstream features for the same physical zone collapse to one
|
||||
traffic_events row / gating dedup key."""
|
||||
evt = adapter._parse_feature(make_wzdx_feature(), time.time())
|
||||
d = adapter.to_event(evt).data
|
||||
assert d["external_id"] == "idot-1:WZ-0001"
|
||||
assert d["external_id"] == "wzdx_US-95|43.63|-116.915|lanes reduced, surface work"
|
||||
assert d["source"] == "wzdx"
|
||||
|
||||
|
||||
|
|
@ -259,8 +267,13 @@ def test_tick_fetches_and_populates_events(adapter, monkeypatch):
|
|||
changed = adapter.tick()
|
||||
assert changed is True
|
||||
events = adapter.get_events()
|
||||
# Two DIFFERENT roads (US-95, I-15) -> 2 distinct coalescing keys (not
|
||||
# over-collapsed); same lat/lon centroid + sub_type on both, road differs.
|
||||
assert len(events) == 2
|
||||
assert {e["external_id"] for e in events} == {"idot-1:A", "idot-1:B"}
|
||||
assert {e["external_id"] for e in events} == {
|
||||
"wzdx_US-95|43.63|-116.915|lanes reduced, surface work",
|
||||
"wzdx_I-15|43.63|-116.915|lanes reduced, surface work",
|
||||
}
|
||||
assert adapter.health_status["feed_count"] == 1
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -201,6 +201,117 @@ def test_empty_first_fetch_does_not_consume_seed():
|
|||
assert adapter._firms_seeded is False, "empty fetch must not seed"
|
||||
|
||||
|
||||
# ═════════════════════════════════════════════════════════════════════════════
|
||||
# 2b. Restart-safe cold start: the silent-seed gate keys on the PERSISTED
|
||||
# firms_pixels baseline, NOT just the in-memory _firms_seeded flag (which
|
||||
# resets to False on every process start). A FIRST-EVER run (empty baseline)
|
||||
# still seeds silently; a RESTART with an existing baseline must NOT re-seed
|
||||
# the whole batch — a genuinely-new cluster still broadcasts, and pre-existing
|
||||
# already-stamped pixels never re-burst.
|
||||
# ═════════════════════════════════════════════════════════════════════════════
|
||||
|
||||
def _raw_evt(lat, lon, acq_time, i, acq_date="2026-06-06"):
|
||||
return {
|
||||
"source": "firms", "event_id": f"e{i}", "lat": lat, "lon": lon,
|
||||
"properties": {"frp": 20.0, "confidence": "high",
|
||||
"brightness": 320.0, "acq_date": acq_date,
|
||||
"acq_time": acq_time},
|
||||
}
|
||||
|
||||
|
||||
def test_firms_pixels_empty_helper_reflects_baseline():
|
||||
"""The gate's emptiness probe: True on a fresh DB, False once any pixel is
|
||||
persisted, and True (fail-safe) if the table is missing entirely."""
|
||||
from meshai.config import FIRMSConfig
|
||||
from meshai.env.firms import FIRMSAdapter
|
||||
from meshai.persistence import get_db
|
||||
|
||||
adapter = FIRMSAdapter(FIRMSConfig(map_key="x"))
|
||||
# Fresh isolated DB: table exists (init_db) but holds zero rows -> empty.
|
||||
assert adapter._firms_pixels_empty() is True
|
||||
|
||||
# Persist one pixel via the real ingest path -> baseline no longer empty.
|
||||
_feed(_pixel(lat=43.0, lon=-115.0, acq_time="1200"), now=1780728000,
|
||||
seed=True)
|
||||
assert _pixel_count() == 1
|
||||
assert adapter._firms_pixels_empty() is False
|
||||
|
||||
# Missing table -> fail-safe treats as empty (never crashes the fetch).
|
||||
get_db().execute("DROP TABLE firms_pixels")
|
||||
assert adapter._firms_pixels_empty() is True
|
||||
|
||||
|
||||
def test_first_ever_run_empty_baseline_seeds_silently():
|
||||
"""FIRST-EVER run: empty firms_pixels + _firms_seeded False -> a ≥3-pixel
|
||||
new cluster SEEDS silently (zero broadcasts) and the baseline is written."""
|
||||
from meshai.config import FIRMSConfig
|
||||
from meshai.env.firms import FIRMSAdapter
|
||||
|
||||
adapter = FIRMSAdapter(FIRMSConfig(map_key="x"))
|
||||
assert adapter._firms_seeded is False
|
||||
assert _pixel_count() == 0, "precondition: empty persisted baseline"
|
||||
|
||||
base_lat, base_lon = 43.700, -114.700
|
||||
batch = [_raw_evt(base_lat + 0.001 * i, base_lon, f"12{i:02d}", i)
|
||||
for i in range(4)] # a tight ≥3-pixel cluster
|
||||
out = adapter._run_fusion(batch)
|
||||
|
||||
assert out == [], "first-ever run must silent-seed (zero broadcasts)"
|
||||
assert _pixel_count() == 4, "seeded pixels persisted to the baseline"
|
||||
assert adapter._firms_seeded is True, "first non-empty fetch flips the flag"
|
||||
|
||||
|
||||
def test_restart_with_baseline_new_cluster_broadcasts_no_burst():
|
||||
"""RESTART with an existing baseline: _firms_seeded resets to False, but the
|
||||
persisted firms_pixels is NON-empty, so cold_start collapses to False. A
|
||||
genuinely-new ≥3-pixel cluster on the post-restart fetch DOES broadcast (not
|
||||
absorbed), and pre-existing already-stamped pixels do NOT re-burst."""
|
||||
from meshai.config import FIRMSConfig
|
||||
from meshai.env.firms import FIRMSAdapter
|
||||
|
||||
# --- Process 1: cold-start seed a cluster (silent), building the baseline. ---
|
||||
a1 = FIRMSAdapter(FIRMSConfig(map_key="x"))
|
||||
old_lat, old_lon = 43.500, -114.500
|
||||
seed_batch = [_raw_evt(old_lat + 0.001 * i, old_lon, f"12{i:02d}", i)
|
||||
for i in range(4)]
|
||||
assert a1._run_fusion(seed_batch) == [], "seed fetch is silent"
|
||||
assert a1._firms_seeded is True
|
||||
seeded_pixels = _pixel_count()
|
||||
assert seeded_pixels == 4
|
||||
stamped_before = _stamped_count()
|
||||
assert stamped_before >= 3, "seeded cluster members are stamped"
|
||||
|
||||
# --- Process 2 (RESTART): the SAME persisted DB, a FRESH adapter whose
|
||||
# in-memory flag reset to False. Baseline is non-empty -> NOT a cold start. ---
|
||||
a2 = FIRMSAdapter(FIRMSConfig(map_key="x"))
|
||||
assert a2._firms_seeded is False, "restart resets the in-memory flag"
|
||||
assert a2._firms_pixels_empty() is False, "but the baseline persists"
|
||||
|
||||
# Post-restart fetch: re-see the SAME pre-existing cluster (no re-burst) PLUS
|
||||
# a genuinely-NEW ≥3-pixel cluster elsewhere (must broadcast).
|
||||
new_lat, new_lon = 44.300, -115.300
|
||||
post_batch = (
|
||||
[_raw_evt(old_lat + 0.001 * i, old_lon, f"12{i:02d}", i)
|
||||
for i in range(4)] # pre-existing -> INSERT-OR-IGNORE no-ops / stamped
|
||||
+ [_raw_evt(new_lat + 0.001 * i, new_lon, f"14{i:02d}", 100 + i)
|
||||
for i in range(3)] # genuinely NEW cluster
|
||||
)
|
||||
out = a2._run_fusion(post_batch)
|
||||
|
||||
cats = [e["properties"].get("category") for e in out]
|
||||
assert "unattributed_hotspot_cluster" in cats, (
|
||||
f"a genuinely-new cluster after restart must broadcast, not be "
|
||||
f"absorbed: {cats}")
|
||||
# Exactly one cluster wire: only the NEW cluster fired; the pre-existing
|
||||
# already-stamped cluster did not re-burst.
|
||||
assert cats.count("unattributed_hotspot_cluster") == 1, (
|
||||
f"pre-existing stamped pixels must not re-burst: {cats}")
|
||||
# The new cluster is centered on the NEW location, not the old one.
|
||||
wire = out[[e["properties"].get("category")
|
||||
for e in out].index("unattributed_hotspot_cluster")]["headline"]
|
||||
assert wire.startswith("🔥 Possible new fire:")
|
||||
|
||||
|
||||
# ═════════════════════════════════════════════════════════════════════════════
|
||||
# 3. Attribution beats clustering (MORA hotspots grow the fire, never cluster)
|
||||
# ═════════════════════════════════════════════════════════════════════════════
|
||||
|
|
|
|||
185
work/tests/test_incident_workzone_suppression.py
Normal file
185
work/tests/test_incident_workzone_suppression.py
Normal file
|
|
@ -0,0 +1,185 @@
|
|||
"""Part 2 tests -- gating/incident.py decide() work-zone broadcast suppression.
|
||||
|
||||
A work zone (source='wzdx', OR any source with sub_type='road_works') must
|
||||
be STORED in traffic_events but its decide() call must return
|
||||
GateResult.broadcast=False. A 511 crash/closure event (different sub_type)
|
||||
must still broadcast unchanged.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from meshai.notifications.gating.incident import decide
|
||||
from meshai.persistence import get_db
|
||||
|
||||
|
||||
_WZDX_DATA = {
|
||||
"external_id": "wzdx_US-95|43.63|-116.915|lanes reduced, surface work",
|
||||
"source": "wzdx",
|
||||
"sub_type": "lanes reduced, surface work",
|
||||
"road": "US-95",
|
||||
"direction": "southbound",
|
||||
"from_loc": None,
|
||||
"to_loc": None,
|
||||
"mile_start": 93,
|
||||
"mile_end": 89,
|
||||
"county": None,
|
||||
"state": None,
|
||||
"lat": 43.63,
|
||||
"lon": -116.915,
|
||||
"impact": "partial",
|
||||
"start_at": 1_783_200_000,
|
||||
"end_at": 1_783_260_000,
|
||||
"magnitude": None,
|
||||
"delay_seconds": None,
|
||||
"icon_category": "road_works",
|
||||
}
|
||||
|
||||
_ITD511_WORKZONE_DATA = {
|
||||
"external_id": "511_WZ-9001",
|
||||
"source": "itd_511",
|
||||
"sub_type": "road_works",
|
||||
"road": "I-84",
|
||||
"direction": "eastbound",
|
||||
"from_loc": None,
|
||||
"to_loc": None,
|
||||
"mile_start": 168,
|
||||
"mile_end": 173,
|
||||
"county": "Ada",
|
||||
"state": "ID",
|
||||
"lat": 43.5,
|
||||
"lon": -116.2,
|
||||
"impact": "partial",
|
||||
"start_at": 1_783_200_000,
|
||||
"end_at": None,
|
||||
"magnitude": None,
|
||||
"delay_seconds": None,
|
||||
"icon_category": "road_works",
|
||||
}
|
||||
|
||||
_ITD511_CRASH_DATA = {
|
||||
"external_id": "511_INC-4242",
|
||||
"source": "itd_511",
|
||||
"sub_type": "accident",
|
||||
"road": "US-93",
|
||||
"direction": "northbound",
|
||||
"from_loc": None,
|
||||
"to_loc": None,
|
||||
"mile_start": 47,
|
||||
"mile_end": None,
|
||||
"county": "Twin Falls",
|
||||
"state": "ID",
|
||||
"lat": 42.5,
|
||||
"lon": -114.5,
|
||||
"impact": None,
|
||||
"start_at": 1_783_200_000,
|
||||
"end_at": None,
|
||||
"magnitude": 3,
|
||||
"delay_seconds": 600,
|
||||
"icon_category": "accident",
|
||||
}
|
||||
|
||||
_ITD511_CLOSURE_DATA = {
|
||||
"external_id": "511_CL-7777",
|
||||
"source": "itd_511",
|
||||
"sub_type": "road_closed",
|
||||
"road": "I-15",
|
||||
"direction": "both",
|
||||
"from_loc": None,
|
||||
"to_loc": None,
|
||||
"mile_start": 100,
|
||||
"mile_end": 102,
|
||||
"county": "Bingham",
|
||||
"state": "ID",
|
||||
"lat": 43.0,
|
||||
"lon": -112.3,
|
||||
"impact": "full_closure",
|
||||
"start_at": 1_783_200_000,
|
||||
"end_at": None,
|
||||
"magnitude": None,
|
||||
"delay_seconds": None,
|
||||
"icon_category": "road_closed",
|
||||
}
|
||||
|
||||
|
||||
def _row(source, external_id):
|
||||
conn = get_db()
|
||||
return conn.execute(
|
||||
"SELECT * FROM traffic_events WHERE source=? AND external_id=?",
|
||||
(source, external_id),
|
||||
).fetchone()
|
||||
|
||||
|
||||
class TestWorkZoneSuppressed:
|
||||
"""A work zone is persisted but never broadcast per-event."""
|
||||
|
||||
def test_native_wzdx_new_row_stored_but_not_broadcast(self):
|
||||
result = decide(dict(_WZDX_DATA), source="wzdx", now=1_783_200_000.0)
|
||||
assert result.broadcast is False
|
||||
assert result.lifecycle == "suppress"
|
||||
assert "work_zone" in result.reason
|
||||
|
||||
row = _row("wzdx", _WZDX_DATA["external_id"])
|
||||
assert row is not None
|
||||
assert row["road"] == "US-95"
|
||||
assert row["last_broadcast_at"] is None
|
||||
|
||||
def test_native_wzdx_existing_row_still_stored_not_broadcast(self):
|
||||
# First decide() inserts the row.
|
||||
decide(dict(_WZDX_DATA), source="wzdx", now=1_783_200_000.0)
|
||||
# Second decide() (existing row branch) must ALSO suppress + persist.
|
||||
updated = dict(_WZDX_DATA, impact="full_closure")
|
||||
result = decide(updated, source="wzdx", now=1_783_200_050.0)
|
||||
assert result.broadcast is False
|
||||
assert result.lifecycle == "suppress"
|
||||
|
||||
row = _row("wzdx", _WZDX_DATA["external_id"])
|
||||
assert row is not None
|
||||
assert row["impact"] == "full_closure" # refreshed by the UPDATE
|
||||
assert row["last_broadcast_at"] is None # never armed
|
||||
|
||||
def test_itd511_road_works_subtype_also_suppressed(self):
|
||||
"""The sub_type='road_works' discriminator catches a work zone
|
||||
arriving through the itd_511 source too (not just native wzdx)."""
|
||||
result = decide(dict(_ITD511_WORKZONE_DATA), source="itd_511",
|
||||
now=1_783_200_000.0)
|
||||
assert result.broadcast is False
|
||||
assert result.lifecycle == "suppress"
|
||||
|
||||
row = _row("itd_511", _ITD511_WORKZONE_DATA["external_id"])
|
||||
assert row is not None
|
||||
assert row["sub_type"] == "road_works"
|
||||
assert row["last_broadcast_at"] is None
|
||||
|
||||
|
||||
class Test511IncidentsStillBroadcast:
|
||||
"""511 crash/closure (non-work-zone sub_types) are completely unaffected."""
|
||||
|
||||
def test_itd511_crash_broadcasts(self):
|
||||
result = decide(dict(_ITD511_CRASH_DATA), source="itd_511",
|
||||
now=1_783_200_000.0)
|
||||
assert result.broadcast is True
|
||||
assert result.lifecycle == "new"
|
||||
assert callable(result.commit)
|
||||
|
||||
row = _row("itd_511", _ITD511_CRASH_DATA["external_id"])
|
||||
assert row is not None
|
||||
|
||||
def test_itd511_closure_broadcasts(self):
|
||||
result = decide(dict(_ITD511_CLOSURE_DATA), source="itd_511",
|
||||
now=1_783_200_000.0)
|
||||
assert result.broadcast is True
|
||||
assert result.lifecycle == "new"
|
||||
assert callable(result.commit)
|
||||
|
||||
row = _row("itd_511", _ITD511_CLOSURE_DATA["external_id"])
|
||||
assert row is not None
|
||||
|
||||
def test_itd511_crash_cold_start_still_broadcasts(self):
|
||||
"""Row exists (from a prior INSERT) but never committed -> cold-start
|
||||
-> still broadcasts (unaffected by the work-zone suppression path)."""
|
||||
decide(dict(_ITD511_CRASH_DATA), source="itd_511", now=1_783_200_000.0)
|
||||
result = decide(dict(_ITD511_CRASH_DATA), source="itd_511",
|
||||
now=1_783_200_010.0)
|
||||
assert result.broadcast is True
|
||||
assert result.lifecycle == "new"
|
||||
180
work/tests/test_wzdx_coalescing.py
Normal file
180
work/tests/test_wzdx_coalescing.py
Normal file
|
|
@ -0,0 +1,180 @@
|
|||
"""Part 1 coalescing tests -- env/wzdx.py external_id merge.
|
||||
|
||||
Multiple WZDx features that resolve to the SAME coalescing key (same road +
|
||||
rounded lat/lon + folded sub_type) must collapse to ONE merged stored event
|
||||
(earliest start_at, latest end_at). Two features on DIFFERENT roads must
|
||||
stay as 2 distinct keys (no over-collapsing).
|
||||
"""
|
||||
|
||||
import time
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
import meshai.central_normalizer as cn
|
||||
from meshai.env.wzdx import WZDxAdapter
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _no_photon(monkeypatch):
|
||||
monkeypatch.setattr(cn, "nearest_town", lambda *a, **k: None)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_config():
|
||||
return SimpleNamespace(
|
||||
enabled=True,
|
||||
feed_source="native",
|
||||
api_key="",
|
||||
base_url="",
|
||||
registry_url="https://datahub.transportation.gov/resource/69qe-yiui.json?$limit=200",
|
||||
registry_ttl=21600,
|
||||
tick_seconds=300,
|
||||
states=["ID"],
|
||||
bbox=[],
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def adapter(mock_config):
|
||||
return WZDxAdapter(mock_config)
|
||||
|
||||
|
||||
def _feature(feat_id, data_source_id, road, direction, start_date, end_date,
|
||||
coordinates, description="Paving operations."):
|
||||
"""Build a WZDx v4 GeoJSON road_event feature (US-26/Gooding-style fans:
|
||||
same road/lat/lon/sub_type, different direction / feat_id / schedule
|
||||
window)."""
|
||||
core = {
|
||||
"event_type": "work-zone",
|
||||
"data_source_id": data_source_id,
|
||||
"road_names": [road],
|
||||
"direction": direction,
|
||||
"description": description,
|
||||
}
|
||||
props = {
|
||||
"core_details": core,
|
||||
"types_of_work": [{"type_name": "surface-work"}],
|
||||
"vehicle_impact": "some-lanes-closed",
|
||||
"start_date": start_date,
|
||||
"end_date": end_date,
|
||||
}
|
||||
return {
|
||||
"id": feat_id,
|
||||
"type": "Feature",
|
||||
"properties": props,
|
||||
"geometry": {"type": "LineString", "coordinates": coordinates},
|
||||
}
|
||||
|
||||
|
||||
def make_feature_collection(*features):
|
||||
return {"type": "FeatureCollection", "features": list(features)}
|
||||
|
||||
|
||||
# Same road/coords (US-26 near Gooding) fanned across direction + feat_id +
|
||||
# schedule-day window -- all 3 features describe the SAME physical zone.
|
||||
_GOODING_COORDS = [[-114.71, 42.94], [-114.72, 42.93]]
|
||||
|
||||
|
||||
def test_same_zone_fan_merges_to_one_event_earliest_start_latest_end(adapter):
|
||||
"""Three US-26/Gooding-style features (same road+lat+lon+sub_type, fanned
|
||||
across direction/feat_id/schedule-day) collapse to ONE merged event_id
|
||||
with the earliest start_at and latest end_at across all three."""
|
||||
feats = [
|
||||
_feature("WZ-100", "idot-1", "US-26", "eastbound",
|
||||
"2026-07-18T06:00:00Z", "2026-07-18T18:00:00Z", _GOODING_COORDS),
|
||||
_feature("WZ-101", "idot-1", "US-26", "eastbound",
|
||||
"2026-07-17T06:00:00Z", "2026-07-19T18:00:00Z", _GOODING_COORDS),
|
||||
_feature("WZ-102", "idot-1", "US-26", "eastbound",
|
||||
"2026-07-19T06:00:00Z", "2026-07-17T18:00:00Z", _GOODING_COORDS),
|
||||
]
|
||||
fc = make_feature_collection(*feats)
|
||||
|
||||
def fake_get(url, timeout=30):
|
||||
return fc
|
||||
|
||||
now = time.time()
|
||||
events = [adapter._parse_feature(f, now) for f in feats]
|
||||
assert all(e is not None for e in events)
|
||||
|
||||
# All three parse to the SAME coalescing key (event_id == external_id).
|
||||
keys = {e["event_id"] for e in events}
|
||||
assert len(keys) == 1, f"expected 1 coalescing key, got {keys}"
|
||||
|
||||
merged = adapter._coalesce_events(events)
|
||||
assert len(merged) == 1
|
||||
|
||||
m = merged[0]
|
||||
# Earliest start_at across the three (2026-07-17T06:00:00Z is earliest).
|
||||
import calendar
|
||||
expected_start = calendar.timegm((2026, 7, 17, 6, 0, 0, 0, 0, 0))
|
||||
expected_end = calendar.timegm((2026, 7, 19, 18, 0, 0, 0, 0, 0))
|
||||
assert m["start_at"] == expected_start
|
||||
assert m["end_at"] == expected_end
|
||||
|
||||
|
||||
def test_different_roads_stay_distinct_keys(adapter):
|
||||
"""Two features on DIFFERENT roads (same lat/lon/sub_type otherwise)
|
||||
must NOT collapse -- 2 distinct coalescing keys survive coalescing."""
|
||||
feat_a = _feature("WZ-200", "idot-1", "US-26", "eastbound",
|
||||
"2026-07-18T06:00:00Z", "2026-07-18T18:00:00Z", _GOODING_COORDS)
|
||||
feat_b = _feature("WZ-201", "idot-1", "I-84", "eastbound",
|
||||
"2026-07-18T06:00:00Z", "2026-07-18T18:00:00Z", _GOODING_COORDS)
|
||||
|
||||
now = time.time()
|
||||
ev_a = adapter._parse_feature(feat_a, now)
|
||||
ev_b = adapter._parse_feature(feat_b, now)
|
||||
assert ev_a is not None and ev_b is not None
|
||||
assert ev_a["event_id"] != ev_b["event_id"]
|
||||
|
||||
merged = adapter._coalesce_events([ev_a, ev_b])
|
||||
assert len(merged) == 2
|
||||
assert {m["event_id"] for m in merged} == {ev_a["event_id"], ev_b["event_id"]}
|
||||
|
||||
|
||||
def test_merge_keeps_none_end_at_when_either_side_open_ended(adapter):
|
||||
"""If EITHER merged feature has no end_date, the merged end_at stays
|
||||
None (an open-ended zone must never get a fabricated end)."""
|
||||
feat_a = _feature("WZ-300", "idot-1", "US-26", "eastbound",
|
||||
"2026-07-18T06:00:00Z", "2026-07-18T18:00:00Z", _GOODING_COORDS)
|
||||
feat_b = dict(feat_a)
|
||||
feat_b["id"] = "WZ-301"
|
||||
feat_b["properties"] = dict(feat_a["properties"])
|
||||
feat_b["properties"]["end_date"] = None
|
||||
|
||||
now = time.time()
|
||||
ev_a = adapter._parse_feature(feat_a, now)
|
||||
ev_b = adapter._parse_feature(feat_b, now)
|
||||
assert ev_a["event_id"] == ev_b["event_id"]
|
||||
assert ev_a["end_at"] is not None
|
||||
assert ev_b["end_at"] is None
|
||||
|
||||
merged = adapter._coalesce_events([ev_a, ev_b])
|
||||
assert len(merged) == 1
|
||||
assert merged[0]["end_at"] is None
|
||||
|
||||
|
||||
def test_fetch_all_end_to_end_coalesces_via_tick(adapter, monkeypatch):
|
||||
"""End-to-end: adapter.tick() -> _fetch_all() coalesces a same-zone fan
|
||||
down to one stored event via the real feed-fetch path."""
|
||||
registry = [{"state": "Idaho", "format": "geojson",
|
||||
"url": "https://itd.idaho.gov/wzdx.geojson"}]
|
||||
feats = [
|
||||
_feature("WZ-400", "idot-1", "US-26", "eastbound",
|
||||
"2026-07-18T06:00:00Z", "2026-07-18T18:00:00Z", _GOODING_COORDS),
|
||||
_feature("WZ-401", "idot-1", "US-26", "westbound",
|
||||
"2026-07-17T06:00:00Z", "2026-07-19T18:00:00Z", _GOODING_COORDS),
|
||||
]
|
||||
fc = make_feature_collection(*feats)
|
||||
|
||||
def fake_get(url, timeout=30):
|
||||
return registry if "datahub" in url else fc
|
||||
|
||||
monkeypatch.setattr(adapter, "_http_get_json", fake_get)
|
||||
changed = adapter.tick()
|
||||
assert changed is True
|
||||
events = adapter.get_events()
|
||||
assert len(events) == 1
|
||||
import calendar
|
||||
assert events[0]["start_at"] == calendar.timegm((2026, 7, 17, 6, 0, 0, 0, 0, 0))
|
||||
assert events[0]["end_at"] == calendar.timegm((2026, 7, 19, 18, 0, 0, 0, 0, 0))
|
||||
314
work/tests/test_wzdx_summary_region_routing.py
Normal file
314
work/tests/test_wzdx_summary_region_routing.py
Normal file
|
|
@ -0,0 +1,314 @@
|
|||
"""Part 3 tests -- wzdx per-region DAILY work-zone count summary.
|
||||
|
||||
WZDxSummaryScheduler.fire_once() must, for each coverage region with >=1
|
||||
active in-coverage work zone, broadcast ONE <=140-char line routed through
|
||||
the region_routes "roads" family cells -- exactly like the fire scheduled
|
||||
path routes through "fire" cells (test_fire_reminder_region_routing.py is
|
||||
the pattern this mirrors). Regions with 0 zones must be skipped. The fire
|
||||
must be on a DAILY clock tick, not change-based (no throttle/dedup table).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import time
|
||||
|
||||
import pytest
|
||||
|
||||
from meshai.config import Config, RegionRouteMatrix
|
||||
from meshai.notifications.pipeline.dispatcher import Dispatcher
|
||||
from meshai.notifications.scheduled.wzdx_summary import WZDxSummaryScheduler
|
||||
from meshai.persistence import get_db
|
||||
|
||||
|
||||
# --------------------------------------------------------------------- recorder
|
||||
|
||||
|
||||
class RecChannel:
|
||||
"""Records each delivery's transport + channel value + message."""
|
||||
|
||||
def __init__(self, rec: list, succeed: bool = True):
|
||||
self.rec = rec
|
||||
self.succeed = succeed
|
||||
|
||||
async def deliver(self, payload, rule):
|
||||
self.rec.append({
|
||||
"delivery_type": rule.delivery_type,
|
||||
"broadcast_channel": getattr(rule, "broadcast_channel", None),
|
||||
"meshcore_channel": getattr(rule, "meshcore_channel", None),
|
||||
"message": payload.message if payload else None,
|
||||
})
|
||||
return self.succeed
|
||||
|
||||
|
||||
# --------------------------------------------------------------------- fixtures
|
||||
|
||||
|
||||
# Same region layout as test_fire_reminder_region_routing.py (SW/SC/East),
|
||||
# but the matrix cells here are keyed under the "roads" family.
|
||||
_COVERAGE_AREAS = [
|
||||
{"name": "SW", "west": -117.0, "south": 43.0, "east": -115.5, "north": 44.2},
|
||||
{"name": "SC", "west": -115.2, "south": 42.0, "east": -113.8, "north": 43.0},
|
||||
{"name": "East", "west": -112.8, "south": 43.0, "east": -111.2, "north": 44.2},
|
||||
]
|
||||
|
||||
_PT = {
|
||||
"SW": (43.6, -116.2),
|
||||
"SC": (42.5, -114.5),
|
||||
"East": (43.5, -112.0),
|
||||
}
|
||||
|
||||
|
||||
def _roads_cfg(*, with_matrix=True, mt_enabled=True, mc_enabled=True,
|
||||
cold_start_grace=0):
|
||||
cfg = Config()
|
||||
cfg.notifications.rules = []
|
||||
cfg.notifications.cold_start_grace_seconds = cold_start_grace
|
||||
|
||||
roads = cfg.notifications.toggles["roads"]
|
||||
roads.enabled = True
|
||||
roads.min_severity = "routine"
|
||||
roads.regions = []
|
||||
roads.freshness_seconds = 0
|
||||
roads.cooldown_seconds = 0
|
||||
roads.broadcast_channel = 9
|
||||
roads.meshcore_channel = "#aida"
|
||||
roads.severity_channels = {
|
||||
"routine": ["mesh_broadcast", "meshcore_broadcast"],
|
||||
"priority": ["mesh_broadcast", "meshcore_broadcast"],
|
||||
"immediate": ["mesh_broadcast", "meshcore_broadcast"],
|
||||
}
|
||||
|
||||
cfg.coverage.enabled = True
|
||||
cfg.coverage.areas = list(_COVERAGE_AREAS)
|
||||
|
||||
if with_matrix:
|
||||
cfg.notifications.region_routes = RegionRouteMatrix(
|
||||
mt_enabled=mt_enabled, mc_enabled=mc_enabled,
|
||||
cells={
|
||||
"roads": {
|
||||
"SW": {"mt": 3, "mc": "#sw-id-aida",
|
||||
"min_severity": "routine", "enabled": True},
|
||||
"SC": {"mt": 2, "mc": "#sc-id-aida",
|
||||
"min_severity": "routine", "enabled": True},
|
||||
"East": {"mt": 5, "mc": "#e-id-aida",
|
||||
"min_severity": "routine", "enabled": True},
|
||||
},
|
||||
},
|
||||
)
|
||||
return cfg
|
||||
|
||||
|
||||
def _dispatcher(cfg, succeed=True):
|
||||
rec: list = []
|
||||
d = Dispatcher(cfg, lambda rule, conn: RecChannel(rec, succeed), connector=None)
|
||||
return d, rec
|
||||
|
||||
|
||||
def _seed_wzdx_row(conn, *, external_id, lat, lon, road="US-95",
|
||||
end_at=None, sub_type="lanes reduced, surface work"):
|
||||
now = int(time.time())
|
||||
conn.execute(
|
||||
"INSERT OR REPLACE INTO traffic_events(source, external_id, road, "
|
||||
"direction, sub_type, impact, lat, lon, first_seen_at, last_seen_at, "
|
||||
"last_broadcast_at, start_at, end_at) "
|
||||
"VALUES ('wzdx', ?, ?, 'southbound', ?, 'partial', ?, ?, ?, ?, NULL, ?, ?)",
|
||||
(external_id, road, sub_type, lat, lon, now, now, now - 3600, end_at),
|
||||
)
|
||||
|
||||
|
||||
def _fire(cfg, succeed=True, now=None):
|
||||
d, rec = _dispatcher(cfg, succeed=succeed)
|
||||
sch = WZDxSummaryScheduler(cfg, d, clock=(lambda: now) if now else time.time)
|
||||
dispatched = asyncio.run(sch.fire_once(now=now))
|
||||
return dispatched, rec
|
||||
|
||||
|
||||
# ============================================================ SW / SC / East
|
||||
|
||||
|
||||
def test_sw_region_summary_routes_ch3_and_sw_mc():
|
||||
"""SW region with 2 active zones -> ONE line, MT ch3 + MC #sw-id-aida."""
|
||||
cfg = _roads_cfg()
|
||||
conn = get_db()
|
||||
lat, lon = _PT["SW"]
|
||||
_seed_wzdx_row(conn, external_id="wz-sw-1", lat=lat, lon=lon, road="US-95")
|
||||
_seed_wzdx_row(conn, external_id="wz-sw-2", lat=lat, lon=lon, road="I-84")
|
||||
|
||||
dispatched, rec = _fire(cfg)
|
||||
assert dispatched == 1
|
||||
|
||||
mt = [r for r in rec if r["delivery_type"] == "mesh_broadcast"]
|
||||
mc = [r for r in rec if r["delivery_type"] == "meshcore_broadcast"]
|
||||
assert len(mt) == 1 and mt[0]["broadcast_channel"] == 3
|
||||
assert len(mc) == 1 and mc[0]["meshcore_channel"] == "#sw-id-aida"
|
||||
|
||||
msg = mt[0]["message"]
|
||||
assert msg.startswith("🚧 SW: 2 active work zones")
|
||||
assert "DM AIDA for details" in msg
|
||||
assert len(msg.encode("utf-8")) <= 140
|
||||
# Must NOT hit the toggle default / rf_propagation-style channels.
|
||||
assert all(r["broadcast_channel"] != 9 for r in mt)
|
||||
|
||||
|
||||
def test_sc_region_summary_routes_ch2_and_sc_mc():
|
||||
cfg = _roads_cfg()
|
||||
conn = get_db()
|
||||
lat, lon = _PT["SC"]
|
||||
_seed_wzdx_row(conn, external_id="wz-sc-1", lat=lat, lon=lon, road="US-93")
|
||||
|
||||
dispatched, rec = _fire(cfg)
|
||||
assert dispatched == 1
|
||||
|
||||
mt = [r for r in rec if r["delivery_type"] == "mesh_broadcast"]
|
||||
mc = [r for r in rec if r["delivery_type"] == "meshcore_broadcast"]
|
||||
assert len(mt) == 1 and mt[0]["broadcast_channel"] == 2
|
||||
assert len(mc) == 1 and mc[0]["meshcore_channel"] == "#sc-id-aida"
|
||||
assert mt[0]["message"].startswith("🚧 SC: 1 active work zones")
|
||||
|
||||
|
||||
def test_east_region_summary_routes_ch5_and_east_mc():
|
||||
cfg = _roads_cfg()
|
||||
conn = get_db()
|
||||
lat, lon = _PT["East"]
|
||||
_seed_wzdx_row(conn, external_id="wz-e-1", lat=lat, lon=lon, road="US-20")
|
||||
|
||||
dispatched, rec = _fire(cfg)
|
||||
assert dispatched == 1
|
||||
|
||||
mt = [r for r in rec if r["delivery_type"] == "mesh_broadcast"]
|
||||
mc = [r for r in rec if r["delivery_type"] == "meshcore_broadcast"]
|
||||
assert len(mt) == 1 and mt[0]["broadcast_channel"] == 5
|
||||
assert len(mc) == 1 and mc[0]["meshcore_channel"] == "#e-id-aida"
|
||||
|
||||
|
||||
# ============================================================ multi-region + skip
|
||||
|
||||
|
||||
def test_multiple_regions_each_get_own_line_zero_zone_regions_skipped():
|
||||
"""SW and East each have zones; SC has none -> exactly 2 dispatched
|
||||
lines (SC skipped, no line emitted for it)."""
|
||||
cfg = _roads_cfg()
|
||||
conn = get_db()
|
||||
sw_lat, sw_lon = _PT["SW"]
|
||||
e_lat, e_lon = _PT["East"]
|
||||
_seed_wzdx_row(conn, external_id="wz-sw-x", lat=sw_lat, lon=sw_lon, road="US-95")
|
||||
_seed_wzdx_row(conn, external_id="wz-e-x", lat=e_lat, lon=e_lon, road="US-20")
|
||||
|
||||
dispatched, rec = _fire(cfg)
|
||||
assert dispatched == 2
|
||||
|
||||
mt = [r for r in rec if r["delivery_type"] == "mesh_broadcast"]
|
||||
channels = {r["broadcast_channel"] for r in mt}
|
||||
assert channels == {3, 5} # SW ch3, East ch5 -- SC (ch2) never fired
|
||||
messages = " ".join(r["message"] for r in mt)
|
||||
assert "SC" not in messages
|
||||
|
||||
|
||||
def test_no_active_zones_dispatches_nothing():
|
||||
cfg = _roads_cfg()
|
||||
dispatched, rec = _fire(cfg)
|
||||
assert dispatched == 0
|
||||
assert rec == []
|
||||
|
||||
|
||||
def test_expired_zone_not_counted():
|
||||
"""A wzdx row whose end_at is in the past is NOT an active zone."""
|
||||
cfg = _roads_cfg()
|
||||
conn = get_db()
|
||||
lat, lon = _PT["SW"]
|
||||
past = int(time.time()) - 86400
|
||||
_seed_wzdx_row(conn, external_id="wz-sw-expired", lat=lat, lon=lon,
|
||||
road="US-95", end_at=past)
|
||||
|
||||
dispatched, rec = _fire(cfg)
|
||||
assert dispatched == 0
|
||||
assert rec == []
|
||||
|
||||
|
||||
def test_open_ended_zone_counted_as_active():
|
||||
"""A wzdx row with end_at=NULL (open-ended) IS an active zone."""
|
||||
cfg = _roads_cfg()
|
||||
conn = get_db()
|
||||
lat, lon = _PT["SW"]
|
||||
_seed_wzdx_row(conn, external_id="wz-sw-open", lat=lat, lon=lon,
|
||||
road="US-95", end_at=None)
|
||||
|
||||
dispatched, rec = _fire(cfg)
|
||||
assert dispatched == 1
|
||||
|
||||
|
||||
# ============================================================ unresolved region
|
||||
|
||||
|
||||
def test_unresolvable_region_falls_back_to_roads_toggle_default():
|
||||
"""A zone whose lat/lon lands in NO named coverage area is simply not
|
||||
counted toward any region (no line emitted for it) -- there IS no
|
||||
'unresolved region' broadcast for a summary line (unlike the fire
|
||||
reminder path, a count needs a NAMED region to attach to)."""
|
||||
cfg = _roads_cfg()
|
||||
conn = get_db()
|
||||
_seed_wzdx_row(conn, external_id="wz-none", lat=0.0, lon=0.0, road="US-1")
|
||||
|
||||
dispatched, rec = _fire(cfg)
|
||||
assert dispatched == 0
|
||||
assert rec == []
|
||||
|
||||
|
||||
def test_region_matched_but_not_in_matrix_falls_back_to_toggle_default():
|
||||
"""A zone in a coverage region (SW) with NO matrix cell for 'roads' in
|
||||
that region falls back to the roads toggle default channels."""
|
||||
cfg = _roads_cfg()
|
||||
del cfg.notifications.region_routes.cells["roads"]["SW"]
|
||||
conn = get_db()
|
||||
lat, lon = _PT["SW"]
|
||||
_seed_wzdx_row(conn, external_id="wz-sw-nomatrix", lat=lat, lon=lon, road="US-95")
|
||||
|
||||
dispatched, rec = _fire(cfg)
|
||||
assert dispatched == 1
|
||||
mt = [r for r in rec if r["delivery_type"] == "mesh_broadcast"]
|
||||
mc = [r for r in rec if r["delivery_type"] == "meshcore_broadcast"]
|
||||
assert len(mt) == 1 and mt[0]["broadcast_channel"] == 9 # roads toggle default
|
||||
assert len(mc) == 1 and mc[0]["meshcore_channel"] == "#aida"
|
||||
|
||||
|
||||
def test_matrix_disabled_falls_back_to_roads_toggle_default():
|
||||
cfg = _roads_cfg(mt_enabled=False, mc_enabled=False)
|
||||
conn = get_db()
|
||||
lat, lon = _PT["SW"]
|
||||
_seed_wzdx_row(conn, external_id="wz-sw-matrixoff", lat=lat, lon=lon, road="US-95")
|
||||
|
||||
dispatched, rec = _fire(cfg)
|
||||
assert dispatched == 1
|
||||
mt = [r for r in rec if r["delivery_type"] == "mesh_broadcast"]
|
||||
mc = [r for r in rec if r["delivery_type"] == "meshcore_broadcast"]
|
||||
assert len(mt) == 1 and mt[0]["broadcast_channel"] == 9
|
||||
assert len(mc) == 1 and mc[0]["meshcore_channel"] == "#aida"
|
||||
|
||||
|
||||
# ============================================================ daily-tick (not change-based)
|
||||
|
||||
|
||||
def test_fires_on_daily_tick_not_change_based():
|
||||
"""fire_once() with the SAME unchanged set of active zones fires again
|
||||
on a second call -- there is deliberately NO change-detection / dedup
|
||||
table gating the daily summary (per spec)."""
|
||||
cfg = _roads_cfg()
|
||||
conn = get_db()
|
||||
lat, lon = _PT["SW"]
|
||||
_seed_wzdx_row(conn, external_id="wz-sw-daily", lat=lat, lon=lon, road="US-95")
|
||||
|
||||
dispatched1, rec1 = _fire(cfg)
|
||||
assert dispatched1 == 1
|
||||
dispatched2, rec2 = _fire(cfg)
|
||||
assert dispatched2 == 1 # fires again -- daily clock tick, not a diff
|
||||
|
||||
|
||||
def test_cold_start_grace_suppresses_first_fire():
|
||||
cfg = _roads_cfg(cold_start_grace=3600)
|
||||
conn = get_db()
|
||||
lat, lon = _PT["SW"]
|
||||
_seed_wzdx_row(conn, external_id="wz-sw-grace", lat=lat, lon=lon, road="US-95")
|
||||
|
||||
dispatched, rec = _fire(cfg)
|
||||
assert dispatched == 0
|
||||
assert rec == []
|
||||
Loading…
Add table
Add a link
Reference in a new issue