From 2eddd9b572a116577282ee03eb9589107639920e Mon Sep 17 00:00:00 2001 From: Matt Johnson Date: Sun, 16 Aug 2026 18:00:12 +0000 Subject: [PATCH 1/3] fix(reminders): resolve Active: fire location via _fire_anchor, not county/state ReminderScheduler._render built the wfigs "Active:" reminder location as a bare " / ".join(county, state), skipping the anchor resolution (geocoder_city -> curated town_anchors -> Photon -> landclass -> county -> state) that the New/Update fire formatter (formatters/fire.py::_fire_anchor) already uses. Reminders now call the same _fire_anchor helper on dict(row) (sqlite3.Row has no .get()) and fit the result to fit_to_budget(..., budget_for("wfigs")), matching how the other renderers guard the mesh packet budget. --- .../notifications/reminders/__init__.py | 8 +- work/tests/test_reminders.py | 91 +++++++++++++++++++ 2 files changed, 97 insertions(+), 2 deletions(-) diff --git a/work/meshai/notifications/reminders/__init__.py b/work/meshai/notifications/reminders/__init__.py index 30b6067..21c93a5 100644 --- a/work/meshai/notifications/reminders/__init__.py +++ b/work/meshai/notifications/reminders/__init__.py @@ -59,6 +59,9 @@ import time from datetime import datetime, timezone from typing import Any, Optional +from meshai.notifications.formatters._budget import budget_for, fit_to_budget +from meshai.notifications.formatters.fire import _fire_anchor + logger = logging.getLogger(__name__) @@ -370,8 +373,9 @@ class ReminderScheduler: cont = row["current_contained_pct"] acres_s = "N/A" if acres is None else f"{int(acres):,} ac" cont_s = "?" if cont is None else f"{int(cont)}%" - anchor = " / ".join(p for p in (row["county"], row["state"]) if p) or "?" - return f"πŸ”₯ {prefix}: {name}, {anchor}: {acres_s}, {cont_s} contained" + anchor = _fire_anchor(dict(row)) + wire = f"πŸ”₯ {prefix}: {name}, {anchor}: {acres_s}, {cont_s} contained" + return fit_to_budget(wire, budget_for("wfigs")) if adapter == "swpc": return f"🌌 {prefix}: ongoing space weather event ({row['event_type']})" if adapter == "itd_511_work_zone": diff --git a/work/tests/test_reminders.py b/work/tests/test_reminders.py index aaa2122..d572a89 100644 --- a/work/tests/test_reminders.py +++ b/work/tests/test_reminders.py @@ -229,3 +229,94 @@ def test_work_zone_reminder_skipped_outside_slot(mock_dispatcher): sch = ReminderScheduler(mock_dispatcher, clock=lambda: now, tick_seconds=60) fired = asyncio.run(sch.tick_once()) assert fired == 0 + + +# ============================================================================ +# Anchor resolution in the wfigs "Active:" reminder wire (_fire_anchor) +# ============================================================================ + + +def test_wfigs_reminder_uses_anchor_not_county(mock_dispatcher): + """A fire near a known curated town_anchors entry (Bliss, ID) renders the + anchor phrase ('mi of Bliss'), not the bare county/state join + the old renderer produced.""" + now = 1_780_000_000 + conn = get_db() + _enable_wfigs_reminders() + # Coordinates ~4 mi NW of Bliss, ID (see _TOWN_ANCHORS_SEED in + # meshai/persistence/curation.py) -- county/state deliberately different + # from "Bliss" so the assertion can't pass by accident via the county. + conn.execute( + "INSERT OR REPLACE INTO fires(irwin_id, incident_name, incident_type, " + "current_acres, current_contained_pct, lat, lon, county, state, " + "declared_at, last_event_at, first_broadcast_at, last_broadcast_at) " + "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)", + ("F-ANCHOR", "Jackalope", "WF", 15, 0, + 42.98, -114.99, "Gooding", "ID", + now - 9 * 3600, now, now - 9 * 3600, now - 9 * 3600), + ) + + sch = ReminderScheduler(mock_dispatcher, clock=lambda: now) + fired = asyncio.run(sch.tick_once()) + assert fired == 1 + args = mock_dispatcher.dispatch_scheduled_fire_broadcast.call_args.kwargs + text = args["text"] + assert "Bliss" in text + assert "of Bliss" in text + # Old format ("Gooding / ID") must be gone. + assert "Gooding / ID" not in text + + +def test_wfigs_reminder_falls_back_when_no_anchor_nearby(mock_dispatcher): + """A fire with no lat/lon at all still renders sensibly via + _fire_anchor's fallback chain (county -> state -> "location unknown"), + and never raises.""" + now = 1_780_000_000 + conn = get_db() + _enable_wfigs_reminders() + conn.execute( + "INSERT OR REPLACE INTO fires(irwin_id, incident_name, incident_type, " + "current_acres, current_contained_pct, lat, lon, county, state, " + "declared_at, last_event_at, first_broadcast_at, last_broadcast_at) " + "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)", + ("F-NOANCHOR", "Remote Fire", "WF", 42, 5, + None, None, "Owyhee", "ID", + now - 9 * 3600, now, now - 9 * 3600, now - 9 * 3600), + ) + + sch = ReminderScheduler(mock_dispatcher, clock=lambda: now) + fired = asyncio.run(sch.tick_once()) + assert fired == 1 + args = mock_dispatcher.dispatch_scheduled_fire_broadcast.call_args.kwargs + text = args["text"] + # Falls through landclass (absent) to the "{county} Co {state}" tier. + assert "Owyhee Co ID" in text + assert "Active" in text + + +def test_wfigs_reminder_truncates_to_budget(mock_dispatcher): + """A very long incident name + anchor is truncated to the wfigs budget + (fit_to_budget) rather than exceeding the mesh packet limit.""" + now = 1_780_000_000 + conn = get_db() + _enable_wfigs_reminders() + long_name = ("The Extremely Long Wildfire Incident Name That Keeps Going " + "And Going And Going Well Past Any Reasonable Mesh Packet Budget") + conn.execute( + "INSERT OR REPLACE INTO fires(irwin_id, incident_name, incident_type, " + "current_acres, current_contained_pct, lat, lon, county, state, " + "declared_at, last_event_at, first_broadcast_at, last_broadcast_at) " + "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)", + ("F-LONG", long_name, "WF", 123456, 3, + 42.98, -114.99, "Gooding", "ID", + now - 9 * 3600, now, now - 9 * 3600, now - 9 * 3600), + ) + + from meshai.notifications.formatters._budget import budget_for + sch = ReminderScheduler(mock_dispatcher, clock=lambda: now) + fired = asyncio.run(sch.tick_once()) + assert fired == 1 + args = mock_dispatcher.dispatch_scheduled_fire_broadcast.call_args.kwargs + text = args["text"] + assert len(text) <= budget_for("wfigs") + assert text.endswith("…") From c0572b04a191c06040eab7e1c01b959052b1f5cc Mon Sep 17 00:00:00 2001 From: Matt Johnson Date: Sun, 16 Aug 2026 18:20:59 +0000 Subject: [PATCH 2/3] feat: user-crafted scheduled announcements (custom_announcements) Free-text broadcasts the owner types straight into the GUI on a clock-slot schedule (daily / interval_days / weekly / monthly) -- no placeholders, no data sources, no templating, no SQL from the user. - v30 migration: custom_announcements table (own explicit channel list per row, new rows start disabled). - CustomAnnouncementScheduler (notifications/scheduled/custom_announcements.py): 60s tick modelled on ReminderScheduler; monthly day-of-month clamped via calendar.monthrange; restart-safe dedup keyed on the local calendar date of last_sent_at; spacing_seconds roll-call pacing between announcements firing in the same tick. - Dispatcher.dispatch_scheduled_custom_broadcast(): delivers to the announcement's own channel list (no toggle/region_routes matrix), one mesh_broadcasts_out audit row per target, cold-start grace. - New announcement_routes.py router: GET/POST/PUT/DELETE /api/announcements + POST .../preview (wire text + char/byte count, never sends). No send-now endpoint anywhere. - Wired into notifications/pipeline/__init__.py (alongside ReminderScheduler) and dashboard/server.py; single_packet_max_chars runtime override added in main.py's budget list. 57 new tests (recurrence kinds, Feb-29/28 clamp, restart-safe dedup, pacing, budget truncation, full input validation, multi-target audit rows, no send-anywhere guarantee). Full suite: 2090 passed, 0 failed (up from 2033 baseline), 9 pre-existing warnings. Co-Authored-By: Claude Opus 5 (1M context) --- .../dashboard/api/announcement_routes.py | 372 ++++++++++++++++++ work/meshai/dashboard/server.py | 2 + work/meshai/main.py | 3 +- .../meshai/notifications/pipeline/__init__.py | 22 ++ .../notifications/pipeline/dispatcher.py | 144 +++++++ .../scheduled/custom_announcements.py | 353 +++++++++++++++++ work/meshai/persistence/migrations/v30.sql | 56 +++ work/tests/test_announcement_routes.py | 324 +++++++++++++++ .../test_custom_announcement_dispatcher.py | 198 ++++++++++ .../test_custom_announcements_scheduler.py | 357 +++++++++++++++++ 10 files changed, 1830 insertions(+), 1 deletion(-) create mode 100644 work/meshai/dashboard/api/announcement_routes.py create mode 100644 work/meshai/notifications/scheduled/custom_announcements.py create mode 100644 work/meshai/persistence/migrations/v30.sql create mode 100644 work/tests/test_announcement_routes.py create mode 100644 work/tests/test_custom_announcement_dispatcher.py create mode 100644 work/tests/test_custom_announcements_scheduler.py diff --git a/work/meshai/dashboard/api/announcement_routes.py b/work/meshai/dashboard/api/announcement_routes.py new file mode 100644 index 0000000..1c8917d --- /dev/null +++ b/work/meshai/dashboard/api/announcement_routes.py @@ -0,0 +1,372 @@ +"""Custom scheduled-announcement REST API (v30 custom_announcements table). + +New resource type (owner-authored free-text broadcasts on a clock-slot +schedule -- see notifications/scheduled/custom_announcements.py), so this is +a NEW router file rather than an extension of notification_routes.py: +notification_routes.py's CRUD-shaped endpoints (region-routing, rules) are +all config-file-backed (save_section/load_config over YAML), while +announcements are individual SQLite rows with their own id, closer in shape +to adapter_config_routes.py's raw get_db()/HTTPException style -- which this +file follows (grouped list, per-row GET/PUT/DELETE, explicit 400s with a +clear message per bad field, no ORM). + +Endpoints: + GET /api/announcements -- list all, newest first + POST /api/announcements -- create (ALWAYS starts disabled) + PUT /api/announcements/{id} -- partial update (validated as a whole) + DELETE /api/announcements/{id} -- delete + POST /api/announcements/{id}/preview -- exact wire text + char/byte + count against budget. + NEVER sends. + +There is deliberately NO "send now" / test-send route anywhere in this file. +""" +from __future__ import annotations + +import json +import logging +import time +from typing import Any, Dict, List, Optional + +from fastapi import APIRouter, HTTPException, Request +from pydantic import BaseModel + +from meshai.notifications.formatters._budget import budget_for, fit_to_budget + +logger = logging.getLogger(__name__) + +router = APIRouter(tags=["announcements"]) + +_VALID_KINDS = {"daily", "interval_days", "weekly", "monthly"} +_VALID_TRANSPORTS = {"meshtastic", "meshcore"} + + +# ============================================================================ +# request bodies +# ============================================================================ + + +class AnnouncementCreateBody(BaseModel): + name: str + message: str + schedule_kind: str + time_of_day: str + interval_days: Optional[int] = None + dow_mask: Optional[List[Any]] = None + day_of_month: Optional[int] = None + timezone: str = "America/Boise" + channels: List[Dict[str, Any]] + + +class AnnouncementUpdateBody(BaseModel): + """Partial update -- only fields present in the request change. + + The MERGED result (existing row + these overrides) is re-validated as a + whole, so e.g. switching schedule_kind to "weekly" without also + supplying dow_mask is rejected even though dow_mask itself wasn't + touched by this request. + """ + name: Optional[str] = None + message: Optional[str] = None + schedule_kind: Optional[str] = None + time_of_day: Optional[str] = None + interval_days: Optional[int] = None + dow_mask: Optional[List[Any]] = None + day_of_month: Optional[int] = None + timezone: Optional[str] = None + channels: Optional[List[Dict[str, Any]]] = None + enabled: Optional[bool] = None + + +# ============================================================================ +# helpers +# ============================================================================ + + +def _get_conn(): + from meshai.persistence import get_db + return get_db() + + +def _row_to_dict(r) -> dict: + channels_raw = r["channels"] + try: + channels = json.loads(channels_raw) if channels_raw else [] + except Exception: + channels = [] + dow_mask_raw = r["dow_mask"] + try: + dow_mask = json.loads(dow_mask_raw) if dow_mask_raw else None + except Exception: + dow_mask = None + return { + "announcement_id": r["announcement_id"], + "name": r["name"], + "message": r["message"], + "schedule_kind": r["schedule_kind"], + "time_of_day": r["time_of_day"], + "interval_days": r["interval_days"], + "dow_mask": dow_mask, + "day_of_month": r["day_of_month"], + "timezone": r["timezone"], + "channels": channels, + "enabled": bool(r["enabled"]), + "last_sent_at": r["last_sent_at"], + "created_at": r["created_at"], + "updated_at": r["updated_at"], + } + + +def _validate_time_of_day(value: Any) -> str: + if not isinstance(value, str) or ":" not in value: + raise HTTPException(400, f"time_of_day must be 'HH:MM', got {value!r}") + parts = value.split(":") + if len(parts) != 2: + raise HTTPException(400, f"time_of_day must be 'HH:MM', got {value!r}") + try: + hh, mm = int(parts[0]), int(parts[1]) + except ValueError: + raise HTTPException(400, f"time_of_day must be 'HH:MM' with numeric parts, got {value!r}") + if not (0 <= hh <= 23 and 0 <= mm <= 59): + raise HTTPException(400, f"time_of_day out of range, got {value!r}") + return f"{hh:02d}:{mm:02d}" + + +def _validate_timezone(value: Any) -> str: + if not isinstance(value, str) or not value.strip(): + raise HTTPException(400, "timezone must be a non-empty string") + try: + from zoneinfo import ZoneInfo + ZoneInfo(value) + except Exception: + raise HTTPException(400, f"unknown timezone {value!r}") + return value + + +def _validate_channels(value: Any) -> List[Dict[str, Any]]: + """SHAPE validation only. Rejects an empty list (nowhere to send) but + places NO upper bound on length -- any number of mixed meshtastic/ + meshcore targets is valid. Does NOT check whether a channel is + currently configured on a live radio (that is a send-time concern -- + dispatch_scheduled_custom_broadcast logs a warning and skips a target + that no longer exists rather than failing the whole announcement).""" + if not isinstance(value, list) or len(value) == 0: + raise HTTPException(400, "channels must be a non-empty list") + out = [] + for i, item in enumerate(value): + if not isinstance(item, dict): + raise HTTPException(400, f"channels[{i}] must be an object") + transport = item.get("transport") + if transport not in _VALID_TRANSPORTS: + raise HTTPException( + 400, + f"channels[{i}].transport must be one of {sorted(_VALID_TRANSPORTS)}, " + f"got {transport!r}", + ) + channel = item.get("channel") + if transport == "meshtastic": + if isinstance(channel, bool) or not isinstance(channel, int): + raise HTTPException( + 400, f"channels[{i}].channel must be an int meshtastic " + f"channel index, got {channel!r}") + else: # meshcore + if not isinstance(channel, str) or not channel.strip(): + raise HTTPException( + 400, f"channels[{i}].channel must be a non-empty " + f"meshcore channel name, got {channel!r}") + entry = {"transport": transport, "channel": channel} + name = item.get("name") + if name is not None: + entry["name"] = name + out.append(entry) + return out + + +def _validate_dow_mask(value: Any) -> List[bool]: + if not isinstance(value, list) or len(value) != 7 or not all( + isinstance(b, bool) for b in value + ): + raise HTTPException( + 400, f"dow_mask must be a list of exactly 7 booleans, got {value!r}") + return value + + +def _validate_full(merged: dict) -> dict: + """Validate a full (post-merge) announcement record. Returns a cleaned + dict ready to persist. Raises HTTPException(400) on the first bad field.""" + name = merged.get("name") + if not isinstance(name, str) or not name.strip(): + raise HTTPException(400, "name must be a non-empty string") + + message = merged.get("message") + if not isinstance(message, str) or not message.strip(): + raise HTTPException(400, "message must be a non-empty string") + + kind = merged.get("schedule_kind") + if kind not in _VALID_KINDS: + raise HTTPException( + 400, f"schedule_kind must be one of {sorted(_VALID_KINDS)}, got {kind!r}") + + time_of_day = _validate_time_of_day(merged.get("time_of_day")) + tz_name = _validate_timezone(merged.get("timezone") or "America/Boise") + channels = _validate_channels(merged.get("channels")) + + interval_days = merged.get("interval_days") + dow_mask = merged.get("dow_mask") + day_of_month = merged.get("day_of_month") + + if kind == "interval_days": + if not isinstance(interval_days, int) or isinstance(interval_days, bool) or interval_days < 1: + raise HTTPException( + 400, f"interval_days must be an int >= 1 for schedule_kind=" + f"'interval_days', got {interval_days!r}") + dow_mask = None + day_of_month = None + elif kind == "weekly": + dow_mask = _validate_dow_mask(dow_mask) + interval_days = None + day_of_month = None + elif kind == "monthly": + if (not isinstance(day_of_month, int) or isinstance(day_of_month, bool) + or not (1 <= day_of_month <= 31)): + raise HTTPException( + 400, f"day_of_month must be an int 1-31 for schedule_kind=" + f"'monthly', got {day_of_month!r}") + interval_days = None + dow_mask = None + else: # daily + interval_days = None + dow_mask = None + day_of_month = None + + return { + "name": name.strip(), + "message": message, + "schedule_kind": kind, + "time_of_day": time_of_day, + "interval_days": interval_days, + "dow_mask": json.dumps(dow_mask) if dow_mask is not None else None, + "day_of_month": day_of_month, + "timezone": tz_name, + "channels": json.dumps(channels), + } + + +# ============================================================================ +# endpoints +# ============================================================================ + + +@router.get("/announcements") +async def list_announcements(request: Request) -> list[dict]: + conn = _get_conn() + rows = conn.execute( + "SELECT * FROM custom_announcements ORDER BY announcement_id DESC" + ).fetchall() + return [_row_to_dict(r) for r in rows] + + +@router.post("/announcements") +async def create_announcement(request: Request, body: AnnouncementCreateBody) -> dict: + """Create a new announcement. ALWAYS starts disabled (enabled=0) -- + the owner arms it explicitly via PUT after reviewing a preview. This + endpoint never sends anything.""" + clean = _validate_full(body.model_dump()) + now = time.time() + conn = _get_conn() + cur = conn.execute( + "INSERT INTO custom_announcements " + "(name, message, schedule_kind, time_of_day, interval_days, " + "dow_mask, day_of_month, timezone, channels, enabled, " + "last_sent_at, created_at, updated_at) " + "VALUES (?,?,?,?,?,?,?,?,?,0,NULL,?,?)", + (clean["name"], clean["message"], clean["schedule_kind"], + clean["time_of_day"], clean["interval_days"], clean["dow_mask"], + clean["day_of_month"], clean["timezone"], clean["channels"], + now, now), + ) + new_id = cur.lastrowid + logger.info("announcement created id=%s name=%r (disabled)", new_id, clean["name"]) + r = conn.execute( + "SELECT * FROM custom_announcements WHERE announcement_id=?", (new_id,) + ).fetchone() + return _row_to_dict(r) + + +@router.put("/announcements/{announcement_id}") +async def update_announcement( + announcement_id: int, request: Request, body: AnnouncementUpdateBody +) -> dict: + conn = _get_conn() + existing = conn.execute( + "SELECT * FROM custom_announcements WHERE announcement_id=?", + (announcement_id,), + ).fetchone() + if existing is None: + raise HTTPException(404, f"announcement {announcement_id} not found") + + current = _row_to_dict(existing) + overrides = {k: v for k, v in body.model_dump().items() if v is not None} + enabled_override = overrides.pop("enabled", None) + merged = {**current, **overrides} + clean = _validate_full(merged) + + new_enabled = current["enabled"] if enabled_override is None else bool(enabled_override) + + now = time.time() + conn.execute( + "UPDATE custom_announcements SET name=?, message=?, schedule_kind=?, " + "time_of_day=?, interval_days=?, dow_mask=?, day_of_month=?, " + "timezone=?, channels=?, enabled=?, updated_at=? " + "WHERE announcement_id=?", + (clean["name"], clean["message"], clean["schedule_kind"], + clean["time_of_day"], clean["interval_days"], clean["dow_mask"], + clean["day_of_month"], clean["timezone"], clean["channels"], + 1 if new_enabled else 0, now, announcement_id), + ) + logger.info("announcement updated id=%s enabled=%s", announcement_id, new_enabled) + r = conn.execute( + "SELECT * FROM custom_announcements WHERE announcement_id=?", + (announcement_id,), + ).fetchone() + return _row_to_dict(r) + + +@router.delete("/announcements/{announcement_id}") +async def delete_announcement(announcement_id: int, request: Request) -> dict: + conn = _get_conn() + existing = conn.execute( + "SELECT announcement_id FROM custom_announcements WHERE announcement_id=?", + (announcement_id,), + ).fetchone() + if existing is None: + raise HTTPException(404, f"announcement {announcement_id} not found") + conn.execute( + "DELETE FROM custom_announcements WHERE announcement_id=?", + (announcement_id,), + ) + logger.info("announcement deleted id=%s", announcement_id) + return {"ok": True, "deleted": announcement_id} + + +@router.post("/announcements/{announcement_id}/preview") +async def preview_announcement(announcement_id: int, request: Request) -> dict: + """Return the exact wire text and its size against the mesh packet + budget. Does NOT send -- there is no send-now endpoint in this API.""" + conn = _get_conn() + r = conn.execute( + "SELECT message FROM custom_announcements WHERE announcement_id=?", + (announcement_id,), + ).fetchone() + if r is None: + raise HTTPException(404, f"announcement {announcement_id} not found") + + budget = budget_for("custom_announcements") + wire = fit_to_budget(r["message"] or "", budget) + return { + "wire_text": wire, + "char_count": len(wire), + "byte_count": len(wire.encode("utf-8")), + "budget": budget, + "truncated": wire != (r["message"] or ""), + } diff --git a/work/meshai/dashboard/server.py b/work/meshai/dashboard/server.py index b81dda6..2a3ce09 100644 --- a/work/meshai/dashboard/server.py +++ b/work/meshai/dashboard/server.py @@ -61,6 +61,7 @@ def create_app() -> FastAPI: from .api.debug_routes import router as debug_router from .api.serial_ports_routes import router as serial_ports_router from .api.gauge_sites_import import router as gauge_sites_import_router + from .api.announcement_routes import router as announcement_router app.include_router(system_router, prefix="/api") app.include_router(serial_ports_router, prefix="/api") @@ -77,6 +78,7 @@ def create_app() -> FastAPI: app.include_router(notification_router, prefix="/api") app.include_router(secrets_router, prefix="/api") app.include_router(debug_router, prefix="/api") + app.include_router(announcement_router, prefix="/api") # WebSocket router (no prefix, path is /ws/live) app.include_router(ws_router) diff --git a/work/meshai/main.py b/work/meshai/main.py index cffc049..c7b02e6 100644 --- a/work/meshai/main.py +++ b/work/meshai/main.py @@ -469,7 +469,8 @@ class MeshAI: # MUST match the adapter_config section each handler reads its budget from # (see meshai.central.budget.budget_for calls in the handlers). from meshai.adapter_config import set_runtime_override - for _adapter in ("nws", "incident", "wfigs", "avalanche", "satpass", "usgs_quake"): + for _adapter in ("nws", "incident", "wfigs", "avalanche", "satpass", "usgs_quake", + "custom_announcements"): set_runtime_override(_adapter, "single_packet_max_chars", self.connector.max_chars) # Passive mesh context buffer diff --git a/work/meshai/notifications/pipeline/__init__.py b/work/meshai/notifications/pipeline/__init__.py index 2fb8fe9..4a81ad2 100644 --- a/work/meshai/notifications/pipeline/__init__.py +++ b/work/meshai/notifications/pipeline/__init__.py @@ -43,6 +43,12 @@ try: from meshai.notifications.reminders import ReminderScheduler except ImportError: ReminderScheduler = None +try: + from meshai.notifications.scheduled.custom_announcements import ( + CustomAnnouncementScheduler, + ) +except ImportError: + CustomAnnouncementScheduler = None from meshai.notifications.pipeline.inhibitor import Inhibitor from meshai.notifications.pipeline.grouper import Grouper from meshai.notifications.pipeline.toggle_filter import ToggleFilter @@ -288,6 +294,22 @@ async def start_pipeline(bus: EventBus, config) -> DigestScheduler: _lg.getLogger("meshai.pipeline").exception( "reminder scheduler failed to start") + # Custom announcements scheduler -- runs alongside the reminder + # scheduler. Best-effort: failures must NOT break notifications + # pipeline startup. + if CustomAnnouncementScheduler is not None: + try: + comps = getattr(bus, "_pipeline_components", {}) or {} + disp = comps.get("dispatcher") + if disp is not None: + ca_sched = CustomAnnouncementScheduler(disp) + await ca_sched.start() + comps["custom_announcement_scheduler"] = ca_sched + bus._pipeline_components = comps + except Exception: + import logging as _lg + _lg.getLogger("meshai.pipeline").exception( + "custom announcement scheduler failed to start") # Phase 2.16.1: periodically flush the grouper so coalesced events are # delivered within the window even when poll cadence is sparse. diff --git a/work/meshai/notifications/pipeline/dispatcher.py b/work/meshai/notifications/pipeline/dispatcher.py index 3034af2..72a19a5 100644 --- a/work/meshai/notifications/pipeline/dispatcher.py +++ b/work/meshai/notifications/pipeline/dispatcher.py @@ -1330,6 +1330,150 @@ class Dispatcher: ch_type) return delivered_any + async def dispatch_scheduled_custom_broadcast( + self, text: str, *, + announcement_id: int, + slot_key: str, + channels: list, + ) -> bool: + """Scheduled broadcast for a user-authored custom announcement. + + Unlike dispatch_scheduled_fire_broadcast / dispatch_scheduled_roads_ + broadcast (which resolve channels via a toggle + region_routes + matrix), a custom announcement carries its OWN explicit destination + list -- set by the owner in the GUI, never derived from a toggle or + a region. Every entry in `channels` is delivered independently and + gets its OWN mesh_broadcasts_out audit row: an announcement with N + mixed Meshtastic/MeshCore targets produces N delivery attempts and N + audit rows. There is no cap on N and no "primary channel" concept. + + channels: list of {"transport": "meshtastic"|"meshcore", + "channel": |, + "name": }. + A target with an unrecognised transport is logged and skipped. A + target whose channel is valid in SHAPE but no longer configured on + the live radio is NOT a hard failure -- create_channel()/deliver() + already logs a warning and returns False for that case (missing + connector, unset meshcore_channel, no meshcore transport, etc.); this + method adds its own warning on top and moves on to the next target, + so one stale MeshCore channel never aborts the rest of the + announcement. + + Fan-out pacing: sends are issued one target at a time, same as + dispatch_scheduled_fire_broadcast's `plan` loop. Each send goes + through create_channel() -> connector.send_message_async() -> + RadioSendQueue, which already applies that transport's inter-packet + jitter -- so multi-target announcements are naturally paced without + a second spacing mechanism here. + + source_event_table is always "custom_announcements"; + source_event_pk is f"{announcement_id}:{slot_key}" so the audit + trail carries a key unique per (announcement, fired slot) rather + than colliding across sends of the same announcement on different + days. + + Cold-start grace still applies (consistent with the other scheduled + broadcasts). Returns True on at least one successful mesh delivery. + """ + 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 custom broadcast " + "id=%s slot=%s", announcement_id, slot_key) + return False + + if not channels: + self._logger.info( + "scheduled-custom-broadcast: no channels for id=%s; dropping", + announcement_id) + return False + + from meshai.notifications.events import make_event, make_payload_from_event + from types import SimpleNamespace + ev = make_event( + source="custom_announcement", category="custom_announcement", + severity="routine", title=text, + ) + ev.data["_meshai_precomposed"] = True + # Minimal stand-in for a toggle object -- _toggle_to_rule() only + # reads .name off it when mt_override/mc_override are both supplied + # (which they always are here, one or the other per target). + fake_tog = SimpleNamespace(name=f"custom_announcement:{announcement_id}") + + source_event_pk = f"{announcement_id}:{slot_key}" + delivered_any = False + for target in channels: + target = target or {} + transport = target.get("transport") + chan = target.get("channel") + if transport == "meshtastic": + ch_type = "mesh_broadcast" + elif transport == "meshcore": + ch_type = "meshcore_broadcast" + else: + self._logger.warning( + "scheduled-custom-broadcast: unknown transport %r for " + "id=%s; skipping target", transport, announcement_id) + continue + + rule = self._toggle_to_rule( + fake_tog, ch_type, ev, + mt_override=(chan if ch_type == "mesh_broadcast" else None), + mc_override=(chan 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-custom-broadcast: delivery raised for id=%s " + "transport=%s channel=%r", + announcement_id, transport, chan) + success = False + + if success: + delivered_any = True + self._logger.info( + "scheduled-custom-broadcast: dispatched id=%s slot=%s " + "via %s ch=%r", announcement_id, slot_key, ch_type, chan) + else: + self._logger.warning( + "scheduled-custom-broadcast: delivery failed/skipped " + "for id=%s transport=%s channel=%r (channel may not be " + "currently configured on the radio)", + announcement_id, transport, chan) + + # One audit row PER TARGET, mirroring the fire/roads scheduled + # broadcasts' per-channel audit insert. + try: + from meshai.persistence import get_db + conn = get_db() + bytes_sent = len(text.encode("utf-8")) if text else 0 + audit_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, + "custom_announcements", source_event_pk, bytes_sent, 0, + audit_transport, 1 if success else 0), + ) + except Exception: + self._logger.exception( + "scheduled-custom-broadcast: audit row insert failed " + "for id=%s transport=%s", announcement_id, transport) + 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/custom_announcements.py b/work/meshai/notifications/scheduled/custom_announcements.py new file mode 100644 index 0000000..c817efe --- /dev/null +++ b/work/meshai/notifications/scheduled/custom_announcements.py @@ -0,0 +1,353 @@ +"""CustomAnnouncementScheduler -- user-crafted scheduled announcements. + +The owner's words: "at xx:xx time every day / other day / specific day / +week / month, send the following message: 'literally anything I type in +this box in the GUI'". Free text only -- NO placeholders, NO data sources, +NO templating. + +Modelled on ReminderScheduler's clock-slot tick loop +(notifications/reminders/__init__.py) -- same 60s tick, same timezone +localisation via `zoneinfo`, same `spacing_seconds` roll-call pacing between +consecutive dispatches -- but reading from the `custom_announcements` table +(persistence/migrations/v30.sql) instead of adapter_config, and dispatching +via ``Dispatcher.dispatch_scheduled_custom_broadcast()`` (the announcement's +OWN explicit channel list) instead of a toggle/region_routes path. + +Recurrence kinds (schedule_kind column) +---------------------------------------- + daily -- fires every day at time_of_day. + interval_days -- fires every N days, counted from the LOCAL CALENDAR + DATE (in the announcement's own timezone) of + created_at. created_at's date is day 0; day N, 2N, 3N... + are eligible. Documented anchor choice: created_at + rather than "now" so the cadence is stable and doesn't + drift if the scheduler is down for a while -- an + announcement created on day 0 with interval_days=2 + always lands on days 0, 2, 4, ... relative to its own + creation date, never renumbered by a restart. + weekly -- fires on days where dow_mask[local_weekday] is True. + dow_mask is Mon-first (index 0 = Monday .. 6 = Sunday), + matching Python's datetime.weekday(). + monthly -- fires on day_of_month, CLAMPED to the last day of the + local month (calendar.monthrange) so day_of_month=31 + fires on Feb 28 (or 29 in a leap year), Apr 30, etc. + +Dedup (restart-safe, no double-send on a slot) +----------------------------------------------- +Every schedule_kind fires AT MOST ONCE per local calendar day (the +recurrence math above decides WHICH days are eligible; time_of_day is the +only slot on an eligible day). That makes "the concrete local datetime of +the slot" reducible to just the local CALENDAR DATE in the announcement's +own timezone -- there is never a second slot to disambiguate within a day. + +The dedup key is therefore (announcement_id, local_date_str). It is checked +by comparing the LOCAL DATE of the stored `last_sent_at` (converted into the +announcement's timezone) against the local date of the slot under +consideration: if they match, the slot has already fired and is skipped. +`last_sent_at` is written through to SQLite immediately after a successful +send (before moving to the next row), so a restart mid-minute re-reads the +same persisted date on its next tick and will not re-fire a slot that +already went out -- there is no in-memory-only state a crash can lose. + +(The remaining crash window -- a process dying in between a successful +`dispatch_scheduled_custom_broadcast()` call returning True and the +`UPDATE ... SET last_sent_at` write landing -- mirrors the exact same +after-dispatch-stamp pattern already used by ReminderScheduler and the +other scheduled dispatchers in this codebase; closing it fully would need a +durable pre-commit intent log, which nothing else here has either.) + +Pacing +------ +Reuses ReminderScheduler's `_space()` mechanism verbatim in spirit: a +shared `_last_dispatch_at` timestamp and a minimum gap between consecutive +SUCCESSFUL announcement dispatches, so N announcements eligible in the same +tick go out spaced apart rather than bursting. This is roll-call-level +pacing (announcement vs. announcement). Fan-out to the multiple channels +WITHIN one announcement is not separately throttled here -- each channel +send already goes through `RadioSendQueue`'s per-transport inter-packet +jitter at the connector layer (the same mechanism dispatch_scheduled_ +fire_broadcast's multi-channel `plan` loop relies on), so a 5-target mixed +announcement is naturally paced without inventing a second mechanism. +""" +from __future__ import annotations + +import asyncio +import calendar +import json +import logging +import time +from datetime import datetime, timezone +from typing import Any, Optional + +try: + from zoneinfo import ZoneInfo +except ImportError: # pragma: no cover (3.9+ only) + ZoneInfo = None + +from meshai.notifications.formatters._budget import budget_for, fit_to_budget + +logger = logging.getLogger(__name__) + + +_TICK_SECONDS = 60.0 + +# Minimum gap between consecutive SUCCESSFUL announcement dispatches when +# more than one is eligible in the same tick. Matches the FirePacer / +# ReminderScheduler default so every scheduled-broadcast exit paces at the +# same rate. +_DEFAULT_SPACING_SECONDS = 60.0 + +_VALID_KINDS = frozenset({"daily", "interval_days", "weekly", "monthly"}) + + +# ============================================================================ +# Pure helpers -- no DB / dispatcher, easy to unit test directly. +# ============================================================================ + + +def clamp_day_of_month(year: int, month: int, day: int) -> int: + """Clamp `day` (1-31) to the last real day of `year`-`month`. + + 31 -> 28 or 29 in February, 30 in Apr/Jun/Sep/Nov, unchanged elsewhere. + """ + last = calendar.monthrange(year, month)[1] + return min(max(1, int(day)), last) + + +def _localize(now: float, tz_name: str) -> datetime: + dt_utc = datetime.fromtimestamp(now, tz=timezone.utc) + if ZoneInfo is not None: + try: + return dt_utc.astimezone(ZoneInfo(tz_name)) + except Exception: + logger.warning("custom_announcements: bad timezone %r; using UTC", tz_name) + return dt_utc + + +def _local_date_str(epoch: float, tz_name: str) -> str: + return _localize(epoch, tz_name).strftime("%Y-%m-%d") + + +def is_day_eligible(row: dict, local_dt: datetime) -> bool: + """Does `local_dt`'s local calendar date match this announcement's + recurrence pattern? (Ignores time-of-day -- caller checks that + separately.) `row` is a mapping with the custom_announcements columns. + """ + kind = row["schedule_kind"] + if kind == "daily": + return True + + if kind == "interval_days": + interval = int(row["interval_days"] or 1) + if interval <= 0: + interval = 1 + tz_name = row["timezone"] or "America/Boise" + created_at = row["created_at"] + if created_at is None: + return False + anchor_date = _localize(float(created_at), tz_name).date() + delta_days = (local_dt.date() - anchor_date).days + return delta_days >= 0 and delta_days % interval == 0 + + if kind == "weekly": + mask = row["dow_mask"] + if isinstance(mask, str): + try: + mask = json.loads(mask) + except Exception: + mask = None + if not isinstance(mask, list) or len(mask) != 7: + logger.warning( + "custom_announcements: bad dow_mask %r for weekly " + "announcement; treating as never-fire", mask) + return False + return bool(mask[local_dt.weekday()]) + + if kind == "monthly": + dom = row["day_of_month"] + if dom is None: + return False + clamped = clamp_day_of_month(local_dt.year, local_dt.month, int(dom)) + return local_dt.day == clamped + + logger.warning("custom_announcements: unknown schedule_kind=%r", kind) + return False + + +# ============================================================================ +# Scheduler +# ============================================================================ + + +class CustomAnnouncementScheduler: + """Fires enabled custom_announcements rows at their configured slots.""" + + def __init__(self, dispatcher, *, + clock=None, sleep=None, + tick_seconds: float = _TICK_SECONDS, + spacing_seconds: float = _DEFAULT_SPACING_SECONDS): + self._dispatcher = dispatcher + self._clock = clock or time.time + self._sleep = sleep or asyncio.sleep + self._tick = tick_seconds + self._spacing = spacing_seconds + self._task: Optional[asyncio.Task] = None + self._stop: Optional[asyncio.Event] = None + # Shared across every announcement -- the mesh is one shared medium, + # same design as ReminderScheduler._last_dispatch_at. + self._last_dispatch_at: Optional[float] = None + + async def start(self) -> None: + if self._task is not None and not self._task.done(): + raise RuntimeError("CustomAnnouncementScheduler already running") + self._stop = asyncio.Event() + self._task = asyncio.create_task(self._run(), name="custom-announcement-scheduler") + logger.info("CustomAnnouncementScheduler started; tick=%ss", self._tick) + + async def stop(self) -> None: + if self._stop: self._stop.set() + if self._task: + try: await self._task + except Exception: pass + + async def _run(self) -> None: + while not (self._stop and self._stop.is_set()): + try: + await self.tick_once() + except Exception: + logger.exception("CustomAnnouncementScheduler tick failed") + try: + await asyncio.wait_for(self._stop.wait(), timeout=self._tick) + return + except asyncio.TimeoutError: + pass + + async def tick_once(self, now: Optional[float] = None) -> int: + """One pass over every enabled announcement. Returns count fired. + + Public so tests can drive single ticks deterministically.""" + now = now if now is not None else self._clock() + try: + from meshai.persistence import get_db + conn = get_db() + except Exception: + return 0 + + rows = conn.execute( + "SELECT * FROM custom_announcements WHERE enabled = 1" + ).fetchall() + if not rows: + return 0 + + fired = 0 + for row in rows: + try: + if await self._maybe_fire(dict(row), now): + fired += 1 + except Exception: + logger.exception( + "custom_announcements: tick failed for id=%s", + row["announcement_id"]) + return fired + + # ---- per-row ---------------------------------------------------- + + async def _maybe_fire(self, row: dict, now: float) -> bool: + tz_name = row["timezone"] or "America/Boise" + local_dt = _localize(now, tz_name) + + if not is_day_eligible(row, local_dt): + return False + + hh_mm = row["time_of_day"] or "" + try: + hh, mm = hh_mm.split(":") + slot_min = int(hh) * 60 + int(mm) + except Exception: + logger.warning( + "custom_announcements: bad time_of_day=%r for id=%s", + hh_mm, row["announcement_id"]) + return False + + current_min = local_dt.hour * 60 + local_dt.minute + tick_min = max(1, int(self._tick / 60)) + # Has the slot just passed (within the last tick)? Mirrors + # ReminderScheduler._tick_clock's window check. + if not (current_min - tick_min <= slot_min <= current_min): + return False + + # Dedup: already sent today (this local calendar date)? + today_str = local_dt.strftime("%Y-%m-%d") + last_sent_at = row.get("last_sent_at") + if last_sent_at is not None: + if _local_date_str(float(last_sent_at), tz_name) == today_str: + return False + + channels = row["channels"] + if isinstance(channels, str): + try: + channels = json.loads(channels) + except Exception: + channels = [] + if not channels: + logger.warning( + "custom_announcements: id=%s has no channels; skipping", + row["announcement_id"]) + return False + + message = row["message"] or "" + wire = fit_to_budget(message, budget_for("custom_announcements")) + if not wire: + return False + + if not await self._space(): + return False # stop() signalled -- abandon the roll-call + + slot_key = f"{today_str}T{hh_mm}" + ok = False + try: + ok = bool(await self._dispatcher.dispatch_scheduled_custom_broadcast( + text=wire, + announcement_id=row["announcement_id"], + slot_key=slot_key, + channels=channels, + )) + except Exception: + logger.exception( + "custom_announcements: dispatch failed id=%s slot=%s", + row["announcement_id"], slot_key) + + if ok: + self._stamp_sent(row["announcement_id"], now) + self._last_dispatch_at = self._clock() + return ok + + async def _space(self) -> bool: + """Wait out the inter-announcement pacing gap. Returns False if + stop() fired while waiting (caller abandons the roll-call). Mirrors + ReminderScheduler._space() exactly.""" + if self._spacing <= 0 or self._last_dispatch_at is None: + return True + remaining = self._spacing - (self._clock() - self._last_dispatch_at) + if remaining <= 0: + return True + if self._stop is not None: + try: + await asyncio.wait_for(self._stop.wait(), timeout=remaining) + return False + except asyncio.TimeoutError: + return True + await self._sleep(remaining) + return True + + def _stamp_sent(self, announcement_id: int, now: float) -> None: + try: + from meshai.persistence import get_db + conn = get_db() + except Exception: + return + conn.execute( + "UPDATE custom_announcements SET last_sent_at=?, updated_at=? " + "WHERE announcement_id=?", + (now, now, announcement_id), + ) diff --git a/work/meshai/persistence/migrations/v30.sql b/work/meshai/persistence/migrations/v30.sql new file mode 100644 index 0000000..e9cf4e3 --- /dev/null +++ b/work/meshai/persistence/migrations/v30.sql @@ -0,0 +1,56 @@ +-- v30 custom_announcements: user-crafted scheduled announcements. +-- +-- Free-text broadcasts the owner types straight into the GUI ("at 08:00 +-- every day, send: ") -- NO placeholders, NO templating, +-- NO data source. Distinct from every other scheduled broadcast in this +-- codebase (band_conditions, reminders, wzdx_summary) in that it carries its +-- OWN explicit destination list (`channels`) instead of routing through a +-- toggle + region_routes matrix -- see dispatch_scheduled_custom_broadcast() +-- in notifications/pipeline/dispatcher.py. +-- +-- schedule_kind determines which of the cadence columns is authoritative: +-- daily -- time_of_day only. +-- interval_days -- time_of_day + interval_days (anchored at created_at's +-- local calendar date -- see CustomAnnouncementScheduler). +-- weekly -- time_of_day + dow_mask (JSON list of 7 booleans, +-- Mon-first, i.e. index 0 = Monday .. 6 = Sunday). +-- monthly -- time_of_day + day_of_month (1-31; clamped to the last +-- day of shorter months at fire time, e.g. 31 fires on +-- Feb 28/29). +-- +-- channels is a JSON list of {"transport": "meshtastic"|"meshcore", +-- "channel": |, "name": } -- +-- UNBOUNDED length, arbitrary mix of transports, no "primary channel" +-- concept. Every entry gets its own delivery + its own mesh_broadcasts_out +-- audit row (dispatch_scheduled_custom_broadcast loops the list). +-- +-- New announcements start DISABLED (enabled=0) -- the owner arms them +-- explicitly after reviewing via POST /api/announcements/{id}/preview. +-- There is deliberately NO "send now" endpoint anywhere in this feature. +-- +-- last_sent_at doubles as the scheduler's restart-safe dedup marker: it is +-- stamped with the actual send epoch, and the scheduler compares the LOCAL +-- CALENDAR DATE (in `timezone`) of last_sent_at against the local date of +-- the slot under consideration -- since every schedule_kind fires at most +-- once per local calendar day, same-date == already-sent-today, and this +-- survives a restart because it is written through to SQLite immediately +-- after a successful send (see CustomAnnouncementScheduler._maybe_fire). + +CREATE TABLE IF NOT EXISTS custom_announcements ( + announcement_id INTEGER PRIMARY KEY AUTOINCREMENT, + name TEXT NOT NULL, + message TEXT NOT NULL, + schedule_kind TEXT NOT NULL, -- daily | interval_days | weekly | monthly + time_of_day TEXT NOT NULL, -- "HH:MM" + interval_days INTEGER, -- interval_days: N (2 = every other day) + dow_mask TEXT, -- weekly: JSON list of 7 booleans, Mon-first + day_of_month INTEGER, -- monthly: 1-31, clamped at fire time + timezone TEXT NOT NULL DEFAULT 'America/Boise', + channels TEXT NOT NULL, -- JSON list of {"transport","channel","name"?} + enabled INTEGER NOT NULL DEFAULT 0, + last_sent_at REAL, + created_at REAL NOT NULL, + updated_at REAL NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_custom_announcements_enabled ON custom_announcements(enabled); +CREATE INDEX IF NOT EXISTS idx_custom_announcements_kind ON custom_announcements(schedule_kind); diff --git a/work/tests/test_announcement_routes.py b/work/tests/test_announcement_routes.py new file mode 100644 index 0000000..823f697 --- /dev/null +++ b/work/tests/test_announcement_routes.py @@ -0,0 +1,324 @@ +"""API tests for /api/announcements (custom scheduled announcements). + +Uses FastAPI TestClient, mirroring tests/test_adapter_config_api.py. +Covers CRUD, validation of every bad-input case named in the spec, the +create-always-disabled invariant, and that no code path under this router +ever sends anything (preview is preview-only; there is no send-now route). +""" +from __future__ import annotations + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from meshai.dashboard.api.announcement_routes import router + + +@pytest.fixture +def client(): + app = FastAPI() + app.include_router(router, prefix="/api") + return TestClient(app) + + +def _valid_body(**overrides): + body = { + "name": "Morning greeting", + "message": "Good morning, mesh!", + "schedule_kind": "daily", + "time_of_day": "08:00", + "channels": [{"transport": "meshtastic", "channel": 2}], + } + body.update(overrides) + return body + + +# ============================================================================ +# Create -- always starts disabled +# ============================================================================ + + +def test_create_returns_201_shape_and_starts_disabled(client): + r = client.post("/api/announcements", json=_valid_body()) + assert r.status_code == 200 + body = r.json() + assert body["enabled"] is False + assert body["name"] == "Morning greeting" + assert body["announcement_id"] is not None + + +def test_create_ignores_client_supplied_enabled_true(client): + """There is no field in AnnouncementCreateBody for enabled at all -- + a client cannot arm an announcement at creation time.""" + r = client.post("/api/announcements", json={**_valid_body(), "enabled": True}) + assert r.status_code == 200 + assert r.json()["enabled"] is False + + +# ============================================================================ +# List / get / delete +# ============================================================================ + + +def test_list_returns_created_rows(client): + client.post("/api/announcements", json=_valid_body(name="A")) + client.post("/api/announcements", json=_valid_body(name="B")) + r = client.get("/api/announcements") + assert r.status_code == 200 + names = {row["name"] for row in r.json()} + assert names == {"A", "B"} + + +def test_delete_removes_row(client): + created = client.post("/api/announcements", json=_valid_body()).json() + aid = created["announcement_id"] + r = client.delete(f"/api/announcements/{aid}") + assert r.status_code == 200 + assert r.json()["ok"] is True + assert client.get("/api/announcements").json() == [] + + +def test_delete_unknown_id_404s(client): + r = client.delete("/api/announcements/99999") + assert r.status_code == 404 + + +def test_update_unknown_id_404s(client): + r = client.put("/api/announcements/99999", json={"name": "x"}) + assert r.status_code == 404 + + +# ============================================================================ +# Update -- can arm (enable) an announcement, partial merge re-validates whole +# ============================================================================ + + +def test_update_can_enable_after_review(client): + created = client.post("/api/announcements", json=_valid_body()).json() + aid = created["announcement_id"] + r = client.put(f"/api/announcements/{aid}", json={"enabled": True}) + assert r.status_code == 200 + assert r.json()["enabled"] is True + + +def test_update_switching_kind_without_required_field_is_rejected(client): + """Switching to weekly without supplying dow_mask must fail even though + dow_mask itself wasn't part of this request -- the MERGED record is + what gets validated.""" + created = client.post("/api/announcements", json=_valid_body()).json() + aid = created["announcement_id"] + r = client.put(f"/api/announcements/{aid}", json={"schedule_kind": "weekly"}) + assert r.status_code == 400 + + +def test_update_message_only_preserves_other_fields(client): + created = client.post("/api/announcements", json=_valid_body()).json() + aid = created["announcement_id"] + r = client.put(f"/api/announcements/{aid}", json={"message": "updated text"}) + assert r.status_code == 200 + body = r.json() + assert body["message"] == "updated text" + assert body["schedule_kind"] == "daily" + assert body["channels"] == [{"transport": "meshtastic", "channel": 2}] + + +# ============================================================================ +# Validation -- each bad input named in the spec +# ============================================================================ + + +def test_rejects_empty_message(client): + r = client.post("/api/announcements", json=_valid_body(message="")) + assert r.status_code == 400 + + +def test_rejects_empty_name(client): + r = client.post("/api/announcements", json=_valid_body(name=" ")) + assert r.status_code == 400 + + +def test_rejects_invalid_hh_mm(client): + for bad in ("25:00", "08:60", "not-a-time", "8"): + r = client.post("/api/announcements", json=_valid_body(time_of_day=bad)) + assert r.status_code == 400, f"{bad!r} should have been rejected" + + +def test_rejects_unknown_schedule_kind(client): + r = client.post("/api/announcements", json=_valid_body(schedule_kind="hourly")) + assert r.status_code == 400 + + +def test_rejects_empty_channel_list(client): + r = client.post("/api/announcements", json=_valid_body(channels=[])) + assert r.status_code == 400 + + +def test_does_not_reject_a_long_channel_list(client): + """No cap: a long, mixed-transport channel list is valid.""" + channels = [{"transport": "meshtastic", "channel": i} for i in range(25)] + channels += [{"transport": "meshcore", "channel": f"#ch{i}"} for i in range(25)] + r = client.post("/api/announcements", json=_valid_body(channels=channels)) + assert r.status_code == 200 + assert len(r.json()["channels"]) == 50 + + +def test_rejects_day_of_month_out_of_range(client): + for bad in (0, 32, -1): + r = client.post("/api/announcements", json=_valid_body( + schedule_kind="monthly", day_of_month=bad, + )) + assert r.status_code == 400, f"day_of_month={bad} should have been rejected" + + +def test_monthly_accepts_valid_day_of_month(client): + r = client.post("/api/announcements", json=_valid_body( + schedule_kind="monthly", day_of_month=31, + )) + assert r.status_code == 200 + + +def test_rejects_dow_mask_wrong_length(client): + r = client.post("/api/announcements", json=_valid_body( + schedule_kind="weekly", dow_mask=[True, False, True], + )) + assert r.status_code == 400 + + +def test_rejects_dow_mask_non_boolean_entries(client): + r = client.post("/api/announcements", json=_valid_body( + schedule_kind="weekly", dow_mask=[1, 1, 1, 1, 1, 1, 1], + )) + assert r.status_code == 400 + + +def test_weekly_accepts_valid_dow_mask(client): + r = client.post("/api/announcements", json=_valid_body( + schedule_kind="weekly", dow_mask=[True, False, False, False, False, False, False], + )) + assert r.status_code == 200 + + +def test_rejects_missing_interval_days_for_interval_kind(client): + r = client.post("/api/announcements", json=_valid_body(schedule_kind="interval_days")) + assert r.status_code == 400 + + +def test_interval_days_accepts_valid_value(client): + r = client.post("/api/announcements", json=_valid_body( + schedule_kind="interval_days", interval_days=2, + )) + assert r.status_code == 200 + + +def test_rejects_channel_shape_meshtastic_non_int_channel(client): + r = client.post("/api/announcements", json=_valid_body( + channels=[{"transport": "meshtastic", "channel": "2"}], + )) + assert r.status_code == 400 + + +def test_rejects_channel_shape_meshcore_empty_channel(client): + r = client.post("/api/announcements", json=_valid_body( + channels=[{"transport": "meshcore", "channel": ""}], + )) + assert r.status_code == 400 + + +def test_rejects_channel_shape_unknown_transport(client): + r = client.post("/api/announcements", json=_valid_body( + channels=[{"transport": "carrier_pigeon", "channel": 1}], + )) + assert r.status_code == 400 + + +def test_does_not_hard_reject_meshcore_channel_not_currently_on_radio(client): + """Shape validation only -- a syntactically valid MeshCore channel name + is accepted even though this test never provisions a live radio.""" + r = client.post("/api/announcements", json=_valid_body( + channels=[{"transport": "meshcore", "channel": "#not-provisioned-yet"}], + )) + assert r.status_code == 200 + + +# ============================================================================ +# Preview -- exact wire text + never sends +# ============================================================================ + + +def test_preview_returns_wire_text_and_counts_without_sending(client, monkeypatch): + created = client.post("/api/announcements", json=_valid_body( + message="Good morning, mesh!", + )).json() + aid = created["announcement_id"] + + r = client.post(f"/api/announcements/{aid}/preview") + assert r.status_code == 200 + body = r.json() + assert body["wire_text"] == "Good morning, mesh!" + assert body["char_count"] == len("Good morning, mesh!") + assert body["byte_count"] == len("Good morning, mesh!".encode("utf-8")) + assert body["truncated"] is False + assert "budget" in body + + +def test_preview_truncates_long_message_to_budget(client): + long_message = "y" * 500 + created = client.post("/api/announcements", json=_valid_body( + message=long_message, + )).json() + aid = created["announcement_id"] + + r = client.post(f"/api/announcements/{aid}/preview") + assert r.status_code == 200 + body = r.json() + assert body["char_count"] <= 140 + assert body["truncated"] is True + assert body["wire_text"] != long_message + + +def test_preview_unknown_id_404s(client): + r = client.post("/api/announcements/99999/preview") + assert r.status_code == 404 + + +# ============================================================================ +# No send-anywhere guarantee +# ============================================================================ + + +def test_no_send_endpoint_exists_anywhere_in_router(): + paths = {(getattr(rt, "path", ""), tuple(getattr(rt, "methods", []) or [])) + for rt in router.routes} + for path, methods in paths: + assert "send" not in path.lower(), f"unexpected send-shaped route: {path}" + + +def test_create_does_not_touch_mesh_broadcasts_out(client): + from meshai.persistence import get_db + conn = get_db() + before = conn.execute("SELECT COUNT(*) AS c FROM mesh_broadcasts_out").fetchone()["c"] + client.post("/api/announcements", json=_valid_body()) + after = conn.execute("SELECT COUNT(*) AS c FROM mesh_broadcasts_out").fetchone()["c"] + assert after == before + + +def test_update_does_not_touch_mesh_broadcasts_out(client): + from meshai.persistence import get_db + conn = get_db() + created = client.post("/api/announcements", json=_valid_body()).json() + aid = created["announcement_id"] + before = conn.execute("SELECT COUNT(*) AS c FROM mesh_broadcasts_out").fetchone()["c"] + client.put(f"/api/announcements/{aid}", json={"enabled": True}) + after = conn.execute("SELECT COUNT(*) AS c FROM mesh_broadcasts_out").fetchone()["c"] + assert after == before + + +def test_preview_does_not_touch_mesh_broadcasts_out(client): + from meshai.persistence import get_db + conn = get_db() + created = client.post("/api/announcements", json=_valid_body()).json() + aid = created["announcement_id"] + before = conn.execute("SELECT COUNT(*) AS c FROM mesh_broadcasts_out").fetchone()["c"] + client.post(f"/api/announcements/{aid}/preview") + after = conn.execute("SELECT COUNT(*) AS c FROM mesh_broadcasts_out").fetchone()["c"] + assert after == before diff --git a/work/tests/test_custom_announcement_dispatcher.py b/work/tests/test_custom_announcement_dispatcher.py new file mode 100644 index 0000000..e474075 --- /dev/null +++ b/work/tests/test_custom_announcement_dispatcher.py @@ -0,0 +1,198 @@ +"""Dispatcher.dispatch_scheduled_custom_broadcast -- own explicit channel +list (no toggle / no region_routes matrix), unlimited mixed targets, one +mesh_broadcasts_out audit row per target, cold-start grace. + +Follows the RecChannel recorder pattern from +tests/test_wzdx_summary_region_routing.py. +""" +from __future__ import annotations + +import asyncio + +import pytest + +from meshai.config import Config +from meshai.notifications.pipeline.dispatcher import Dispatcher +from meshai.persistence import get_db + + +class RecChannel: + """Records each delivery's transport + channel value + message.""" + + def __init__(self, rec: list, succeed=True): + self.rec = rec + self.succeed = succeed + + async def deliver(self, payload, rule): + ok = self.succeed(rule) if callable(self.succeed) else self.succeed + 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, + "ok": ok, + }) + return ok + + +def _cfg(cold_start_grace=0): + cfg = Config() + cfg.notifications.rules = [] + cfg.notifications.cold_start_grace_seconds = cold_start_grace + return cfg + + +def _dispatcher(cfg, succeed=True): + rec: list = [] + d = Dispatcher(cfg, lambda rule, conn: RecChannel(rec, succeed), connector=None) + return d, rec + + +def _run(coro): + return asyncio.run(coro) + + +# ============================================================================ +# Multi-target fan-out (any number, mixed transports) +# ============================================================================ + + +def test_five_plus_mixed_targets_all_delivered_with_own_audit_row(): + """5+ mixed targets (3 meshtastic + 2 meshcore) -> every one delivered, + every one gets its own mesh_broadcasts_out row.""" + cfg = _cfg() + d, rec = _dispatcher(cfg, succeed=True) + conn = get_db() + + channels = [ + {"transport": "meshtastic", "channel": 0}, + {"transport": "meshtastic", "channel": 2}, + {"transport": "meshtastic", "channel": 5}, + {"transport": "meshcore", "channel": "#sw-id-aida"}, + {"transport": "meshcore", "channel": "#sc-id-aida"}, + ] + + ok = _run(d.dispatch_scheduled_custom_broadcast( + text="hello everyone", announcement_id=42, slot_key="2026-08-17T08:00", + channels=channels, + )) + assert ok is True + assert len(rec) == 5 + + 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 {r["broadcast_channel"] for r in mt} == {0, 2, 5} + assert {r["meshcore_channel"] for r in mc} == {"#sw-id-aida", "#sc-id-aida"} + assert all(r["message"] == "hello everyone" for r in rec) + + audit_rows = conn.execute( + "SELECT transport, channel, source_event_table, source_event_pk, success " + "FROM mesh_broadcasts_out WHERE source_event_table='custom_announcements' " + "ORDER BY id" + ).fetchall() + assert len(audit_rows) == 5 + for r in audit_rows: + assert r["source_event_table"] == "custom_announcements" + assert r["source_event_pk"] == "42:2026-08-17T08:00" + assert r["success"] == 1 + audit_channels_mt = {r["channel"] for r in audit_rows if r["transport"] == "meshtastic"} + audit_channels_mc = {r["channel"] for r in audit_rows if r["transport"] == "meshcore"} + assert audit_channels_mt == {0, 2, 5} + assert audit_channels_mc == {"#sw-id-aida", "#sc-id-aida"} + + +def test_no_cap_on_number_of_targets(): + """40 mixed targets all get delivered -- no artificial cap anywhere.""" + cfg = _cfg() + d, rec = _dispatcher(cfg, succeed=True) + channels = [{"transport": "meshtastic", "channel": i} for i in range(20)] + channels += [{"transport": "meshcore", "channel": f"#ch{i}"} for i in range(20)] + + ok = _run(d.dispatch_scheduled_custom_broadcast( + text="big fan-out", announcement_id=1, slot_key="s", channels=channels, + )) + assert ok is True + assert len(rec) == 40 + + +def test_empty_channel_list_drops_with_no_delivery(): + cfg = _cfg() + d, rec = _dispatcher(cfg, succeed=True) + ok = _run(d.dispatch_scheduled_custom_broadcast( + text="nowhere to go", announcement_id=1, slot_key="s", channels=[], + )) + assert ok is False + assert rec == [] + + +# ============================================================================ +# A missing/unrecognised target must not abort the rest of the announcement +# ============================================================================ + + +def test_one_failing_target_does_not_abort_the_others(caplog): + """One target 'fails' (e.g. channel no longer configured on the radio -- + deliver() returns False); the other targets still go out, and the + failure still gets its own audit row with success=0.""" + cfg = _cfg() + conn = get_db() + + def succeed_unless_missing(rule): + return getattr(rule, "meshcore_channel", None) != "#gone" + + d, rec = _dispatcher(cfg, succeed=succeed_unless_missing) + channels = [ + {"transport": "meshtastic", "channel": 1}, + {"transport": "meshcore", "channel": "#gone"}, + {"transport": "meshcore", "channel": "#still-here"}, + ] + + ok = _run(d.dispatch_scheduled_custom_broadcast( + text="partial failure ok", announcement_id=7, slot_key="s", + channels=channels, + )) + assert ok is True # at least one delivery succeeded + assert len(rec) == 3 + + audit_rows = conn.execute( + "SELECT channel, success FROM mesh_broadcasts_out " + "WHERE source_event_pk='7:s' ORDER BY id" + ).fetchall() + assert len(audit_rows) == 3 + by_channel = {r["channel"]: r["success"] for r in audit_rows} + assert by_channel["#gone"] == 0 + assert by_channel["#still-here"] == 1 + assert by_channel[1] == 1 + + +def test_unknown_transport_is_skipped_not_fatal(): + cfg = _cfg() + d, rec = _dispatcher(cfg, succeed=True) + channels = [ + {"transport": "carrier_pigeon", "channel": "n/a"}, + {"transport": "meshtastic", "channel": 3}, + ] + + ok = _run(d.dispatch_scheduled_custom_broadcast( + text="skip bad transport", announcement_id=1, slot_key="s", + channels=channels, + )) + assert ok is True + assert len(rec) == 1 + assert rec[0]["broadcast_channel"] == 3 + + +# ============================================================================ +# Cold-start grace (consistent with the other scheduled broadcasts) +# ============================================================================ + + +def test_cold_start_grace_suppresses_first_broadcast(): + cfg = _cfg(cold_start_grace=3600) + d, rec = _dispatcher(cfg, succeed=True) + ok = _run(d.dispatch_scheduled_custom_broadcast( + text="too soon", announcement_id=1, slot_key="s", + channels=[{"transport": "meshtastic", "channel": 0}], + )) + assert ok is False + assert rec == [] diff --git a/work/tests/test_custom_announcements_scheduler.py b/work/tests/test_custom_announcements_scheduler.py new file mode 100644 index 0000000..c10848a --- /dev/null +++ b/work/tests/test_custom_announcements_scheduler.py @@ -0,0 +1,357 @@ +"""CustomAnnouncementScheduler -- recurrence math, dedup, pacing. + +Mirrors the shape of tests/test_fire_reminder_pacing.py and +tests/test_wzdx_summary_region_routing.py: a fake injected clock/sleep so +the suite pays no wall-clock cost, and a mock dispatcher whose +dispatch_scheduled_custom_broadcast is asserted on directly. +""" +from __future__ import annotations + +import asyncio +import json +import time +from datetime import datetime, timezone +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from meshai.notifications.scheduled.custom_announcements import ( + CustomAnnouncementScheduler, + clamp_day_of_month, + is_day_eligible, +) +from meshai.persistence import get_db + + +# ---------- helpers -------------------------------------------------------- + + +_TZ = "America/Boise" # UTC-7 (MST, no DST edge in our fixed test dates) + + +def _epoch_for_local(y, m, d, hh, mm, tz_name=_TZ) -> float: + from zoneinfo import ZoneInfo + dt = datetime(y, m, d, hh, mm, tzinfo=ZoneInfo(tz_name)) + return dt.timestamp() + + +def _insert_announcement( + conn, *, name="Test", message="hello mesh", schedule_kind="daily", + time_of_day="08:00", interval_days=None, dow_mask=None, + day_of_month=None, tz_name=_TZ, channels=None, enabled=1, + created_at=None, last_sent_at=None, +) -> int: + now = created_at if created_at is not None else time.time() + if channels is None: + channels = [{"transport": "meshtastic", "channel": 2}] + cur = conn.execute( + "INSERT INTO custom_announcements " + "(name, message, schedule_kind, time_of_day, interval_days, " + "dow_mask, day_of_month, timezone, channels, enabled, " + "last_sent_at, created_at, updated_at) " + "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)", + (name, message, schedule_kind, time_of_day, interval_days, + json.dumps(dow_mask) if dow_mask is not None else None, + day_of_month, tz_name, json.dumps(channels), enabled, + last_sent_at, now, now), + ) + return cur.lastrowid + + +@pytest.fixture +def mock_dispatcher(): + d = MagicMock() + d.dispatch_scheduled_custom_broadcast = AsyncMock(return_value=True) + return d + + +def _sched(dispatcher, *, clock=None, sleep=None, spacing_seconds=60.0): + return CustomAnnouncementScheduler( + dispatcher, clock=clock, sleep=sleep, spacing_seconds=spacing_seconds, + ) + + +# ============================================================================ +# clamp_day_of_month +# ============================================================================ + + +def test_clamp_day_of_month_february_non_leap(): + assert clamp_day_of_month(2026, 2, 31) == 28 + + +def test_clamp_day_of_month_february_leap(): + assert clamp_day_of_month(2024, 2, 31) == 29 + + +def test_clamp_day_of_month_thirty_day_month(): + assert clamp_day_of_month(2026, 4, 31) == 30 + + +def test_clamp_day_of_month_unaffected_when_in_range(): + assert clamp_day_of_month(2026, 1, 15) == 15 + + +# ============================================================================ +# Recurrence kinds -- each fires on the right day +# ============================================================================ + + +def test_daily_fires_every_day(mock_dispatcher): + conn = get_db() + now = _epoch_for_local(2026, 8, 17, 8, 0) # Monday + _insert_announcement(conn, schedule_kind="daily", time_of_day="08:00") + + fired = asyncio.run(_sched(mock_dispatcher, clock=lambda: now).tick_once()) + assert fired == 1 + + +def test_interval_days_fires_on_multiple_of_interval_from_creation(): + """interval_days=2, created on day 0 -> fires day 0, 2, 4... not day 1, 3.""" + conn = get_db() + anchor = _epoch_for_local(2026, 8, 10, 8, 0) # Monday, day 0 + row_id = _insert_announcement( + conn, schedule_kind="interval_days", interval_days=2, + time_of_day="08:00", created_at=anchor, + ) + row = dict(conn.execute( + "SELECT * FROM custom_announcements WHERE announcement_id=?", (row_id,) + ).fetchone()) + + from zoneinfo import ZoneInfo + day0 = datetime.fromtimestamp(anchor, tz=timezone.utc).astimezone(ZoneInfo(_TZ)) + day1 = day0.replace(day=day0.day + 1) + day2 = day0.replace(day=day0.day + 2) + + assert is_day_eligible(row, day0) is True + assert is_day_eligible(row, day1) is False + assert is_day_eligible(row, day2) is True + + +def test_weekly_fires_only_on_masked_days(): + """dow_mask Mon-first: only Wed (index 2) true.""" + conn = get_db() + mask = [False, False, True, False, False, False, False] + row_id = _insert_announcement( + conn, schedule_kind="weekly", dow_mask=mask, time_of_day="08:00", + ) + row = dict(conn.execute( + "SELECT * FROM custom_announcements WHERE announcement_id=?", (row_id,) + ).fetchone()) + + from zoneinfo import ZoneInfo + tue = datetime(2026, 8, 18, 8, 0, tzinfo=ZoneInfo(_TZ)) # Tuesday + wed = datetime(2026, 8, 19, 8, 0, tzinfo=ZoneInfo(_TZ)) # Wednesday + + assert is_day_eligible(row, tue) is False + assert is_day_eligible(row, wed) is True + + +def test_monthly_fires_on_day_of_month(): + conn = get_db() + row_id = _insert_announcement( + conn, schedule_kind="monthly", day_of_month=15, time_of_day="08:00", + ) + row = dict(conn.execute( + "SELECT * FROM custom_announcements WHERE announcement_id=?", (row_id,) + ).fetchone()) + + from zoneinfo import ZoneInfo + the_14th = datetime(2026, 8, 14, 8, 0, tzinfo=ZoneInfo(_TZ)) + the_15th = datetime(2026, 8, 15, 8, 0, tzinfo=ZoneInfo(_TZ)) + + assert is_day_eligible(row, the_14th) is False + assert is_day_eligible(row, the_15th) is True + + +def test_monthly_day_31_clamps_in_february_non_leap(): + """The headline requirement: day_of_month=31 fires on Feb 28 in a non-leap year.""" + conn = get_db() + row_id = _insert_announcement( + conn, schedule_kind="monthly", day_of_month=31, time_of_day="09:00", + ) + row = dict(conn.execute( + "SELECT * FROM custom_announcements WHERE announcement_id=?", (row_id,) + ).fetchone()) + + from zoneinfo import ZoneInfo + feb28_2026 = datetime(2026, 2, 28, 9, 0, tzinfo=ZoneInfo(_TZ)) # 2026 is not leap + feb27_2026 = datetime(2026, 2, 27, 9, 0, tzinfo=ZoneInfo(_TZ)) + + assert is_day_eligible(row, feb28_2026) is True + assert is_day_eligible(row, feb27_2026) is False + + +def test_monthly_day_31_clamps_in_february_leap_year(): + conn = get_db() + row_id = _insert_announcement( + conn, schedule_kind="monthly", day_of_month=31, time_of_day="09:00", + ) + row = dict(conn.execute( + "SELECT * FROM custom_announcements WHERE announcement_id=?", (row_id,) + ).fetchone()) + + from zoneinfo import ZoneInfo + feb29_2024 = datetime(2024, 2, 29, 9, 0, tzinfo=ZoneInfo(_TZ)) # 2024 IS leap + assert is_day_eligible(row, feb29_2024) is True + + +def test_monthly_day_31_fires_via_full_tick_in_february(): + """End-to-end: tick_once actually dispatches on Feb 28 for day_of_month=31.""" + conn = get_db() + now = _epoch_for_local(2026, 2, 28, 9, 0) + _insert_announcement( + conn, schedule_kind="monthly", day_of_month=31, time_of_day="09:00", + ) + d = MagicMock() + d.dispatch_scheduled_custom_broadcast = AsyncMock(return_value=True) + + fired = asyncio.run(_sched(d, clock=lambda: now).tick_once()) + assert fired == 1 + d.dispatch_scheduled_custom_broadcast.assert_called_once() + + +# ============================================================================ +# Disabled announcements never fire +# ============================================================================ + + +def test_disabled_announcement_never_fires(mock_dispatcher): + conn = get_db() + now = _epoch_for_local(2026, 8, 17, 8, 0) + _insert_announcement(conn, schedule_kind="daily", time_of_day="08:00", enabled=0) + + fired = asyncio.run(_sched(mock_dispatcher, clock=lambda: now).tick_once()) + assert fired == 0 + mock_dispatcher.dispatch_scheduled_custom_broadcast.assert_not_called() + + +# ============================================================================ +# Dedup: never send the same slot twice, restart-safe +# ============================================================================ + + +def test_dedup_prevents_double_send_in_same_slot(mock_dispatcher): + """Two ticks within the same minute window must not double-fire.""" + conn = get_db() + now = _epoch_for_local(2026, 8, 17, 8, 0) + _insert_announcement(conn, schedule_kind="daily", time_of_day="08:00") + + sched = _sched(mock_dispatcher, clock=lambda: now) + fired1 = asyncio.run(sched.tick_once()) + fired2 = asyncio.run(sched.tick_once()) + + assert fired1 == 1 + assert fired2 == 0 + assert mock_dispatcher.dispatch_scheduled_custom_broadcast.call_count == 1 + + +def test_dedup_survives_a_simulated_restart(): + """A brand-new scheduler instance (simulating a process restart) reads + the persisted last_sent_at and still refuses to double-send the slot + that already went out.""" + conn = get_db() + now = _epoch_for_local(2026, 8, 17, 8, 0) + _insert_announcement(conn, schedule_kind="daily", time_of_day="08:00") + + d1 = MagicMock() + d1.dispatch_scheduled_custom_broadcast = AsyncMock(return_value=True) + fired1 = asyncio.run(_sched(d1, clock=lambda: now).tick_once()) + assert fired1 == 1 + + # Simulate restart: a fresh scheduler instance, same DB, ticking again + # a few seconds later within the same slot window. + d2 = MagicMock() + d2.dispatch_scheduled_custom_broadcast = AsyncMock(return_value=True) + fired2 = asyncio.run(_sched(d2, clock=lambda: now + 5).tick_once()) + assert fired2 == 0 + d2.dispatch_scheduled_custom_broadcast.assert_not_called() + + +def test_dedup_allows_the_next_days_slot(mock_dispatcher): + conn = get_db() + day1 = _epoch_for_local(2026, 8, 17, 8, 0) + day2 = _epoch_for_local(2026, 8, 18, 8, 0) + _insert_announcement(conn, schedule_kind="daily", time_of_day="08:00") + + fired1 = asyncio.run(_sched(mock_dispatcher, clock=lambda: day1).tick_once()) + fired2 = asyncio.run(_sched(mock_dispatcher, clock=lambda: day2).tick_once()) + + assert fired1 == 1 + assert fired2 == 1 + + +# ============================================================================ +# Message truncation at the budget +# ============================================================================ + + +def test_message_is_truncated_to_budget(mock_dispatcher): + conn = get_db() + now = _epoch_for_local(2026, 8, 17, 8, 0) + long_message = "x" * 500 + _insert_announcement( + conn, schedule_kind="daily", time_of_day="08:00", message=long_message, + ) + + asyncio.run(_sched(mock_dispatcher, clock=lambda: now).tick_once()) + + call = mock_dispatcher.dispatch_scheduled_custom_broadcast.call_args + sent_text = call.kwargs["text"] + assert len(sent_text) <= 140 + assert sent_text != long_message + + +# ============================================================================ +# Pacing between multiple announcements firing in the same tick +# ============================================================================ + + +class _FakeTime: + def __init__(self, start): + self.now = start + self.sleeps: list[float] = [] + + def clock(self) -> float: + return self.now + + async def sleep(self, seconds: float) -> None: + self.sleeps.append(seconds) + self.now += seconds + + +def test_multiple_announcements_in_one_tick_are_spaced_not_burst(): + conn = get_db() + ft = _FakeTime(_epoch_for_local(2026, 8, 17, 8, 0)) + for i in range(3): + _insert_announcement( + conn, name=f"A{i}", schedule_kind="daily", time_of_day="08:00", + ) + sent_at: list[float] = [] + d = MagicMock() + d.dispatch_scheduled_custom_broadcast = AsyncMock( + side_effect=lambda **kw: sent_at.append(ft.now) or True + ) + + fired = asyncio.run( + _sched(d, clock=ft.clock, sleep=ft.sleep, spacing_seconds=60.0).tick_once() + ) + + assert fired == 3 + gaps = [b - a for a, b in zip(sent_at, sent_at[1:])] + assert gaps == [60.0, 60.0] + + +def test_single_announcement_fire_is_not_delayed(): + conn = get_db() + ft = _FakeTime(_epoch_for_local(2026, 8, 17, 8, 0)) + _insert_announcement(conn, schedule_kind="daily", time_of_day="08:00") + d = MagicMock() + d.dispatch_scheduled_custom_broadcast = AsyncMock(return_value=True) + + fired = asyncio.run( + _sched(d, clock=ft.clock, sleep=ft.sleep, spacing_seconds=60.0).tick_once() + ) + + assert fired == 1 + assert ft.sleeps == [] From c2fd4d7131aa8eaf0ebad783d565535c8cc58a6f Mon Sep 17 00:00:00 2001 From: Matt Johnson Date: Sun, 16 Aug 2026 18:29:26 +0000 Subject: [PATCH 3/3] feat(dashboard): custom-announcements frontend (Announcements panel) Extends the existing ScheduledBroadcasts page with an Announcements section (list + create/edit form) over the announcement_routes.py API from c0572b04. No new page/route/nav entry. - AnnouncementsPanel: list (plain-English schedule summary, channel summary, enabled toggle, last sent, delete-with-confirm) + a create/edit form (message textarea with a live char count against the 140-char budget, schedule-kind-specific controls, Preview button wired to POST /announcements/{id}/preview -- never sends). - AnnouncementChannelPicker: polls GET /api/channels (Meshtastic) and GET /api/meshcore/channels/detail (MeshCore) and renders both as checkbox groups, mixing transports freely. Degrades per radio: shows a plain "unreachable" note and still renders any already-selected channel by its stored name so an existing announcement stays editable even when a radio is down. Blocks save on zero channels. - New announcements always save disabled (enabled is never sent from the create/edit form); the list-row toggle is the only place this form family arms one, matching the backend contract where PUT is the sole place `enabled` can change. - lib/api.ts: fetchMeshtasticChannels + the Announcement CRUD/preview client functions, following the existing fetchJson/detail-surfacing error conventions. Co-Authored-By: Claude Opus 5 (1M context) --- .../components/AnnouncementChannelPicker.tsx | 163 ++++++ .../src/components/AnnouncementsPanel.tsx | 463 ++++++++++++++++++ work/dashboard-frontend/src/lib/api.ts | 117 +++++ .../src/pages/ScheduledBroadcasts.tsx | 5 + 4 files changed, 748 insertions(+) create mode 100644 work/dashboard-frontend/src/components/AnnouncementChannelPicker.tsx create mode 100644 work/dashboard-frontend/src/components/AnnouncementsPanel.tsx diff --git a/work/dashboard-frontend/src/components/AnnouncementChannelPicker.tsx b/work/dashboard-frontend/src/components/AnnouncementChannelPicker.tsx new file mode 100644 index 0000000..2d2b8c7 --- /dev/null +++ b/work/dashboard-frontend/src/components/AnnouncementChannelPicker.tsx @@ -0,0 +1,163 @@ +// Checkbox channel picker for custom announcements. Polls both radios for +// their CURRENT channels and lets the owner check any number of them, +// mixing Meshtastic and MeshCore freely. This is deliberately NOT the +// single-transport ChannelPicker.tsx (that one is Meshtastic-only and +// stores plain channel indices) -- announcement channels are a mixed- +// transport list of {transport, channel, name}. +import { useState, useEffect, useCallback } from 'react' +import { Check, AlertTriangle } from 'lucide-react' +import { + fetchMeshtasticChannels, getMeshcoreChannelsDetail, + type MeshtasticChannel, type MeshcoreChannelsDetail, type AnnouncementChannelRef, +} from '@/lib/api' + +interface Props { + value: AnnouncementChannelRef[] + onChange: (value: AnnouncementChannelRef[]) => void +} + +export default function AnnouncementChannelPicker({ value, onChange }: Props) { + const [mtChannels, setMtChannels] = useState(null) + const [mcDetail, setMcDetail] = useState(null) + const [loading, setLoading] = useState(true) + + const load = useCallback(() => { + setLoading(true) + Promise.allSettled([fetchMeshtasticChannels(), getMeshcoreChannelsDetail()]).then( + ([mt, mc]) => { + setMtChannels(mt.status === 'fulfilled' ? mt.value : []) + setMcDetail(mc.status === 'fulfilled' ? mc.value : { active: false, channels: [] }) + setLoading(false) + } + ) + }, []) + + useEffect(() => { load() }, [load]) + + const isSelected = (transport: 'meshtastic' | 'meshcore', channel: number | string) => + value.some((v) => v.transport === transport && v.channel === channel) + + const toggle = (transport: 'meshtastic' | 'meshcore', channel: number | string, name: string) => { + if (isSelected(transport, channel)) { + onChange(value.filter((v) => !(v.transport === transport && v.channel === channel))) + } else { + onChange([...value, { transport, channel, name }]) + } + } + + if (loading) { + return
Polling radios for current channels...
+ } + + // Meshtastic: only channels the radio currently reports as enabled are + // offerable. Empty list (radio unreachable or nothing enabled) still + // needs to surface any channel this announcement already had selected, + // by its stored name, so an existing announcement stays editable. + const mtLive = (mtChannels ?? []).filter((c) => c.enabled) + const mtStaleSelections = value.filter( + (v) => v.transport === 'meshtastic' && !mtLive.some((c) => c.index === v.channel) + ) + + const mcActive = mcDetail?.active ?? false + const mcLive = mcDetail?.channels ?? [] + const mcStaleSelections = value.filter( + (v) => v.transport === 'meshcore' && !mcLive.some((c) => c.name === v.channel) + ) + + return ( +
+ {/* Meshtastic */} +
+ + {mtLive.length === 0 && ( +
+ + Meshtastic radio unreachable, or it isn't reporting any enabled channels right now. +
+ )} +
+ {mtLive.map((ch) => ( + + ))} + {mtStaleSelections.map((v) => ( + + ))} + {mtLive.length === 0 && mtStaleSelections.length === 0 && ( +
No Meshtastic channels available.
+ )} +
+
+ + {/* MeshCore */} +
+ + {!mcActive && ( +
+ + MeshCore companion unreachable right now. +
+ )} +
+ {mcLive.map((ch) => ( + + ))} + {mcStaleSelections.map((v) => ( + + ))} + {mcLive.length === 0 && mcStaleSelections.length === 0 && ( +
No MeshCore channels available.
+ )} +
+
+ + {value.length === 0 && ( +

Select at least one channel -- an announcement with nowhere to send can't be saved.

+ )} +
+ ) +} + +function CheckBox({ checked }: { checked: boolean }) { + return ( +
+ {checked && } +
+ ) +} diff --git a/work/dashboard-frontend/src/components/AnnouncementsPanel.tsx b/work/dashboard-frontend/src/components/AnnouncementsPanel.tsx new file mode 100644 index 0000000..930bbde --- /dev/null +++ b/work/dashboard-frontend/src/components/AnnouncementsPanel.tsx @@ -0,0 +1,463 @@ +// Custom scheduled announcements (owner-authored free-text broadcasts on a +// clock-slot schedule). CRUD over /api/announcements -- see +// work/meshai/dashboard/api/announcement_routes.py for the backend +// contract this follows exactly (new rows always start disabled; PUT is +// the only place `enabled` can change; preview never sends). +import { useState, useEffect, useCallback } from 'react' +import { Plus, Trash2, Check, X, Loader2, Eye, Megaphone } from 'lucide-react' +import { + fetchAnnouncements, createAnnouncement, updateAnnouncement, deleteAnnouncement, + previewAnnouncement, + type Announcement, type AnnouncementDraft, type AnnouncementChannelRef, type AnnouncementPreview, +} from '@/lib/api' +import { TextInput, NumberInput, TimeInput } from '@/pages/Notifications' +import { SelectInput } from '@/pages/Config' +import AnnouncementChannelPicker from '@/components/AnnouncementChannelPicker' + +const DOW_LABELS = ['Mon', 'Tue', 'Wed', 'Thu', 'Fri', 'Sat', 'Sun'] +const MESSAGE_BUDGET_HINT = 140 // typical mesh packet budget; Preview after saving confirms the exact figure + +function ordinal(n: number): string { + const rem100 = n % 100 + if (rem100 >= 11 && rem100 <= 13) return `${n}th` + switch (n % 10) { + case 1: return `${n}st` + case 2: return `${n}nd` + case 3: return `${n}rd` + default: return `${n}th` + } +} + +function describeSchedule(a: { + schedule_kind: string + time_of_day: string + interval_days: number | null + dow_mask: boolean[] | null + day_of_month: number | null +}): string { + const t = a.time_of_day + switch (a.schedule_kind) { + case 'daily': + return `Every day at ${t}` + case 'interval_days': { + const n = a.interval_days ?? 1 + return `Every ${n} day${n === 1 ? '' : 's'} at ${t}` + } + case 'weekly': { + const days = (a.dow_mask ?? []).map((on, i) => (on ? DOW_LABELS[i] : null)).filter(Boolean) + return days.length ? `${days.join('/')} at ${t}` : `No days selected, at ${t}` + } + case 'monthly': + return `The ${ordinal(a.day_of_month ?? 1)} of each month at ${t}` + default: + return t + } +} + +function channelSummary(channels: AnnouncementChannelRef[]): string { + if (channels.length === 0) return 'No channels' + const names = channels.map((c) => c.name || String(c.channel)) + return `${channels.length} channel${channels.length === 1 ? '' : 's'}: ${names.join(', ')}` +} + +function formatLastSent(v: number | null): string { + if (!v) return 'Never' + return new Date(v * 1000).toLocaleString() +} + +const EMPTY_DRAFT: AnnouncementDraft = { + name: '', + message: '', + schedule_kind: 'daily', + time_of_day: '08:00', + interval_days: 1, + dow_mask: [false, false, false, false, false, false, false], + day_of_month: 1, + timezone: 'America/Boise', + channels: [], +} + +export default function AnnouncementsPanel() { + const [rows, setRows] = useState([]) + const [loading, setLoading] = useState(true) + const [loadError, setLoadError] = useState(null) + + const [formOpen, setFormOpen] = useState(false) + const [editingId, setEditingId] = useState(null) + const [draft, setDraft] = useState(EMPTY_DRAFT) + const [savedMessage, setSavedMessage] = useState('') // the last-persisted message, for the preview staleness note + const [saving, setSaving] = useState(false) + const [saveError, setSaveError] = useState(null) + const [notice, setNotice] = useState(null) + + const [preview, setPreview] = useState(null) + const [previewLoading, setPreviewLoading] = useState(false) + const [previewError, setPreviewError] = useState(null) + + const [busyRowId, setBusyRowId] = useState(null) + + const refresh = useCallback(async () => { + setLoading(true) + setLoadError(null) + try { + setRows(await fetchAnnouncements()) + } catch (e) { + setLoadError(e instanceof Error ? e.message : String(e)) + } finally { + setLoading(false) + } + }, []) + + useEffect(() => { refresh() }, [refresh]) + + const openNew = () => { + setEditingId(null) + setDraft({ ...EMPTY_DRAFT, dow_mask: [...EMPTY_DRAFT.dow_mask!] }) + setSavedMessage('') + setPreview(null) + setPreviewError(null) + setSaveError(null) + setFormOpen(true) + } + + const openEdit = (a: Announcement) => { + setEditingId(a.announcement_id) + setDraft({ + name: a.name, + message: a.message, + schedule_kind: a.schedule_kind, + time_of_day: a.time_of_day, + interval_days: a.interval_days ?? 1, + dow_mask: a.dow_mask ?? [false, false, false, false, false, false, false], + day_of_month: a.day_of_month ?? 1, + timezone: a.timezone, + channels: a.channels, + }) + setSavedMessage(a.message) + setPreview(null) + setPreviewError(null) + setSaveError(null) + setFormOpen(true) + } + + const closeForm = () => { + setFormOpen(false) + setEditingId(null) + setPreview(null) + setPreviewError(null) + setSaveError(null) + } + + const save = async () => { + setSaveError(null) + if (!draft.name.trim()) { setSaveError('Name is required.'); return } + if (!draft.message.trim()) { setSaveError('Message is required.'); return } + if (draft.channels.length === 0) { setSaveError('Select at least one channel.'); return } + if (draft.schedule_kind === 'interval_days' && (!draft.interval_days || draft.interval_days < 1)) { + setSaveError('Interval must be at least 1 day.'); return + } + if (draft.schedule_kind === 'weekly' && !(draft.dow_mask ?? []).some(Boolean)) { + setSaveError('Select at least one day of the week.'); return + } + if (draft.schedule_kind === 'monthly' && (!draft.day_of_month || draft.day_of_month < 1 || draft.day_of_month > 31)) { + setSaveError('Day of month must be between 1 and 31.'); return + } + + setSaving(true) + try { + if (editingId === null) { + await createAnnouncement(draft) + setNotice('Saved. Enable it when you’re ready β€” new announcements always start disabled.') + } else { + // Never send `enabled` from this form -- the list view's toggle is + // the only thing allowed to arm/disarm an announcement. + await updateAnnouncement(editingId, draft) + setNotice('Saved.') + } + setFormOpen(false) + setEditingId(null) + await refresh() + setTimeout(() => setNotice(null), 5000) + } catch (e) { + setSaveError(e instanceof Error ? e.message : String(e)) + } finally { + setSaving(false) + } + } + + const runPreview = async () => { + if (editingId === null) return + setPreviewLoading(true) + setPreviewError(null) + try { + setPreview(await previewAnnouncement(editingId)) + } catch (e) { + setPreviewError(e instanceof Error ? e.message : String(e)) + } finally { + setPreviewLoading(false) + } + } + + const toggleEnabled = async (a: Announcement) => { + setBusyRowId(a.announcement_id) + try { + await updateAnnouncement(a.announcement_id, { enabled: !a.enabled }) + await refresh() + } catch (e) { + alert(`Could not change enabled state: ${e instanceof Error ? e.message : String(e)}`) + } finally { + setBusyRowId(null) + } + } + + const remove = async (a: Announcement) => { + if (!confirm(`Delete announcement "${a.name}"? This cannot be undone.`)) return + setBusyRowId(a.announcement_id) + try { + await deleteAnnouncement(a.announcement_id) + await refresh() + } catch (e) { + alert(`Delete failed: ${e instanceof Error ? e.message : String(e)}`) + setBusyRowId(null) + } + } + + const messageLen = draft.message.length + const overBudget = messageLen > MESSAGE_BUDGET_HINT + const messageDirty = editingId !== null && draft.message !== savedMessage + + return ( +
+
+ + + {rows.length} configured + {!formOpen && ( + + )} +
+

+ Free-text broadcasts you write and schedule yourself, separate from the automatic hazard/weather + notifications above. New announcements always save disabled — nothing transmits until you + switch one on below. +

+ + {notice && ( +
+ {notice} +
+ )} + + {/* ---- Create / edit form ---- */} + {formOpen && ( +
+
+

{editingId === null ? 'New announcement' : 'Edit announcement'}

+ +
+ + {saveError && ( +
{saveError}
+ )} + + setDraft({ ...draft, name: v })} + placeholder="Short label, e.g. Range closure notice" + helper="For your own reference in the list below -- not sent on the air." + /> + +
+ +