diff --git a/work/meshai/adapter_config/defaults.py b/work/meshai/adapter_config/defaults.py index 1ae21a3..b14f9b0 100644 --- a/work/meshai/adapter_config/defaults.py +++ b/work/meshai/adapter_config/defaults.py @@ -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) diff --git a/work/meshai/env/firms.py b/work/meshai/env/firms.py index 2eeb8e3..46c266b 100644 --- a/work/meshai/env/firms.py +++ b/work/meshai/env/firms.py @@ -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 {} diff --git a/work/meshai/env/wzdx.py b/work/meshai/env/wzdx.py index d822e1e..0410487 100644 --- a/work/meshai/env/wzdx.py +++ b/work/meshai/env/wzdx.py @@ -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_|||``) — 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), diff --git a/work/meshai/notifications/env_reporter.py b/work/meshai/notifications/env_reporter.py index 17399b7..cc84636 100644 --- a/work/meshai/notifications/env_reporter.py +++ b/work/meshai/notifications/env_reporter.py @@ -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), diff --git a/work/meshai/notifications/gating/incident.py b/work/meshai/notifications/gating/incident.py index 7babed8..e77fe5d 100644 --- a/work/meshai/notifications/gating/incident.py +++ b/work/meshai/notifications/gating/incident.py @@ -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 ──────────────────── diff --git a/work/meshai/notifications/pipeline/__init__.py b/work/meshai/notifications/pipeline/__init__.py index 074652f..a884835 100644 --- a/work/meshai/notifications/pipeline/__init__.py +++ b/work/meshai/notifications/pipeline/__init__.py @@ -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: diff --git a/work/meshai/notifications/pipeline/dispatcher.py b/work/meshai/notifications/pipeline/dispatcher.py index 5c30e56..6f5430a 100644 --- a/work/meshai/notifications/pipeline/dispatcher.py +++ b/work/meshai/notifications/pipeline/dispatcher.py @@ -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. diff --git a/work/meshai/notifications/scheduled/wzdx_summary.py b/work/meshai/notifications/scheduled/wzdx_summary.py new file mode 100644 index 0000000..11fa1eb --- /dev/null +++ b/work/meshai/notifications/scheduled/wzdx_summary.py @@ -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: + + "🚧 : 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 diff --git a/work/tests/test_adapter_wzdx.py b/work/tests/test_adapter_wzdx.py index d4c4c83..c8d5eb8 100644 --- a/work/tests/test_adapter_wzdx.py +++ b/work/tests/test_adapter_wzdx.py @@ -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 diff --git a/work/tests/test_firms_cluster_f3.py b/work/tests/test_firms_cluster_f3.py index 770c5a2..49c2b6d 100644 --- a/work/tests/test_firms_cluster_f3.py +++ b/work/tests/test_firms_cluster_f3.py @@ -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) # ═════════════════════════════════════════════════════════════════════════════ diff --git a/work/tests/test_incident_workzone_suppression.py b/work/tests/test_incident_workzone_suppression.py new file mode 100644 index 0000000..40df351 --- /dev/null +++ b/work/tests/test_incident_workzone_suppression.py @@ -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" diff --git a/work/tests/test_wzdx_coalescing.py b/work/tests/test_wzdx_coalescing.py new file mode 100644 index 0000000..4362d2e --- /dev/null +++ b/work/tests/test_wzdx_coalescing.py @@ -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)) diff --git a/work/tests/test_wzdx_summary_region_routing.py b/work/tests/test_wzdx_summary_region_routing.py new file mode 100644 index 0000000..14879aa --- /dev/null +++ b/work/tests/test_wzdx_summary_region_routing.py @@ -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 == []