alerts.py

"""
The alert engine: the hub's own voice. Everything screens showed before
this was either scheduled content or an external push (the HA webhook) —
the hub itself could know something worth saying (a real NWS storm
warning, a disk filling up, a screen that stopped heartbeating) and had
no way to put it in front of anyone unless a screen happened to be
assigned the system-alerts dashboard.

Rules are evaluated from scheduler.py every few minutes (quiet hours
INCLUDED — an overnight severe-weather warning is exactly the thing
worth waking a wall of e-paper for; devices polling at the quiet cadence
still pick it up within one poll). Two severities, two behaviors:

  urgent — a real takeover: pushed through the existing overrides store
      to every registered screen for URGENT_TAKEOVER_SECONDS, exactly
      like an HA doorbell push. Deliberately ONCE per alert id (the
      taken-over set persists to alerts_state.json), so a warning that
      stays active for six hours interrupts once and then behaves like a
      notice — screens don't get held hostage by a long-lived warning.
      Today only real NWS warnings (Severe/Extreme) rate this.

  notice — never takes over. While active it time-shares each screen's
      feed: display.resolve() shows the alert slide INJECT_SECONDS out
      of every INJECT_PERIOD_SECONDS, wall-clock anchored — stateless,
      restart-proof, and every screen shows it at the same moment, the
      same principles rotations are built on. When several alerts are
      active, injection windows cycle through them.

Alert sources:
  - NWS active alerts for the hub's location (api.weather.gov — free,
    no key; this also finally makes the alert tile show REAL weather
    warnings instead of the hardcoded demo it launched with).
  - System: disk nearly full, CPU thermally throttled, stalled GUI apps
    (reusing sources.py's checks — same data, now with a delivery path).
  - Fleet: a registered screen that hasn't heartbeated in over an hour
    (shown on the OTHER screens, which is the only place it can be).
  - LocalAutomation: one-shot alert files a separate personal automation
    suite (~/LocalAutomation, not part of this repo) writes to a shared
    queue — weekend art-opening notices, network monitor, and web health
    monitor findings (see _check_local_automation below).

A failed NWS fetch reuses the previous successful result for up to 30
minutes (a network blip shouldn't clear a real storm warning), then
drops it (a stale warning shown forever is worse than none).
"""

import json
import logging
import shutil
import threading
import time
from datetime import datetime
from pathlib import Path

import requests

import config
import db
import overrides
import sources

logger = logging.getLogger("screenhub.alerts")

STATE_PATH = Path(__file__).parent / "alerts_state.json"

# api.weather.gov requires an identifying User-Agent.
_NWS_HEADERS = {
    "User-Agent": "screenhub/1.0 (personal e-paper dashboard)",
    "Accept": "application/geo+json",
}
_NWS_STALE_OK_SECONDS = 1800

_lock = threading.Lock()
_active: dict[str, dict] = {}   # alert_id -> alert dict
_nws_cache: dict = {"fetched_at": 0.0, "alerts": []}


# ---------------------------------------------------------------------------
# Evaluation (scheduler entry point)
# ---------------------------------------------------------------------------

def evaluate(now: float | None = None):
    """Re-derive the active alert set and take over screens for any NEW
    urgent alert. Cheap when nothing's wrong — the common case."""
    now = time.time() if now is None else now
    found = []
    for check in (_check_nws, _check_system, _check_fleet, _check_local_automation):
        try:
            found.extend(check(now))
        except Exception as e:
            logger.warning("alerts: %s failed: %s", check.__name__, e)

    with _lock:
        _active.clear()
        _active.update({a["id"]: a for a in found})

    if found:
        logger.info("alerts: %d active: %s", len(found), [a["id"] for a in found])
    _takeover_new_urgents(found)


def _takeover_new_urgents(found: list[dict]):
    urgents = [a for a in found if a["severity"] == "urgent"]
    if not urgents:
        return
    state = _load_state()
    taken = state.get("taken_over") or []
    for alert in urgents:
        if alert["id"] in taken:
            continue
        dashboard = build_alert_dashboard(alert)
        for device in db.list_devices():
            if device["status"] == "registered":
                overrides.set_override(device["device_id"], dashboard, config.URGENT_TAKEOVER_SECONDS)
        taken.append(alert["id"])
        logger.info("alerts: urgent takeover pushed to fleet: %s", alert["id"])
    # Bounded memory of past takeovers — enough to never re-interrupt for
    # a long-lived warning, small enough to never matter on disk.
    _save_state({"taken_over": taken[-100:]})


# ---------------------------------------------------------------------------
# What display.py reads
# ---------------------------------------------------------------------------

def build_alert_dashboard(alert: dict) -> dict:
    """An inline dashboard (same shape HA's webhook sends) rendering one
    alert through the existing AlertTile — no new tile type needed."""
    return {"layout": "full", "tiles": [{"type": "alert", "config": {
        "title": alert.get("title", "alert"),
        "subtitle": alert.get("subtitle", ""),
        "rows": alert.get("rows") or [],
        "hazard": alert.get("hazard", ""),
        "refresh_seconds": 60,
    }}]}


def current_injection(now: float | None = None) -> dict | None:
    """The alert slide that should time-share a screen's feed right now,
    or None (no active alerts, or outside the injection window). Pure
    wall-clock math — stateless and identical across every screen, like
    rotations."""
    now = time.time() if now is None else now
    with _lock:
        active = sorted(_active.values(), key=lambda a: a["id"])
    if not active:
        return None
    if (now % config.INJECT_PERIOD_SECONDS) >= config.INJECT_SECONDS:
        return None
    index = int(now // config.INJECT_PERIOD_SECONDS) % len(active)
    return build_alert_dashboard(active[index])


def get_status() -> dict:
    """For GET /api/alerts and the dashboard UI's health strip."""
    with _lock:
        active = sorted(_active.values(), key=lambda a: a["id"])
    return {"active": active, "count": len(active)}


# ---------------------------------------------------------------------------
# Rules
# ---------------------------------------------------------------------------

def _check_nws(now: float) -> list[dict]:
    global _nws_cache
    try:
        resp = requests.get(config.NWS_ALERTS_URL, timeout=10, headers=_NWS_HEADERS)
        resp.raise_for_status()
        features = resp.json().get("features") or []
        _nws_cache = {"fetched_at": now, "alerts": [_nws_to_alert(f) for f in features]}
    except Exception as e:
        age = now - _nws_cache["fetched_at"]
        if _nws_cache["alerts"] and age < _NWS_STALE_OK_SECONDS:
            logger.warning("alerts: NWS fetch failed (%s), reusing %.0fs-old result", e, age)
        else:
            logger.warning("alerts: NWS fetch failed (%s), no usable cache", e)
            return []
    return [a for a in _nws_cache["alerts"] if a is not None]


def _nws_to_alert(feature: dict) -> dict | None:
    props = feature.get("properties") or {}
    event = props.get("event") or "Weather Alert"
    severity = props.get("severity") or ""

    ends_iso = props.get("ends") or props.get("expires")
    rows = []
    if ends_iso:
        try:
            ends_local = datetime.fromisoformat(ends_iso).astimezone()
            if ends_local.timestamp() < time.time():
                return None  # already over; /active can briefly lag reality
            rows.append(("until", ends_local.strftime("%a %I:%M %p").replace(" 0", " ")))
        except ValueError:
            pass
    if severity:
        rows.append(("severity", severity))

    # A real WARNING at Severe/Extreme rates a takeover; watches,
    # advisories, and statements just ride the feed.
    urgent = event.lower().endswith("warning") and severity in ("Severe", "Extreme")

    # NWSheadline reads like a banner already ("DAMAGING WINDS EXPECTED
    # THROUGH 8 PM"); the prose headline/description runs to paragraphs
    # and would shrink the banner font into illegibility.
    parameters = props.get("parameters") or {}
    nws_headline = (parameters.get("NWSheadline") or [""])[0]

    return {
        "id": props.get("id") or feature.get("id") or f"nws:{event}",
        "severity": "urgent" if urgent else "notice",
        "title": event,
        "subtitle": props.get("areaDesc") or "",
        "rows": rows,
        "hazard": nws_headline[:120],
    }


def _check_system(now: float) -> list[dict]:
    out = []

    usage = shutil.disk_usage("/")
    pct_used = usage.used / usage.total * 100
    if pct_used >= config.ALERT_DISK_USED_PCT:
        out.append({
            "id": "system:disk-full",
            "severity": "notice",
            "title": "disk almost full",
            "subtitle": "hub boot volume",
            "rows": [("used", f"{pct_used:.0f}%"), ("free", f"{usage.free / 1024**3:.0f} GB")],
            "hazard": "",
        })

    thermal = sources._fetch_thermal()
    limit = thermal.get("value")
    if isinstance(limit, int) and limit < config.ALERT_THERMAL_LIMIT_PCT:
        out.append({
            "id": "system:thermal",
            "severity": "notice",
            "title": "hub running hot",
            "subtitle": "cpu thermally throttled",
            "rows": [("speed limit", f"{limit}%")],
            "hazard": "",
        })

    stalled = [
        item["text"].removesuffix(" not responding")
        for item in sources._check_stalled_apps()
        if item.get("text", "").endswith("not responding")
    ]
    if stalled:
        out.append({
            # One stable id however many apps are stuck — per-app ids
            # would make a second app hang read as a "new" alert.
            "id": "system:stalled-apps",
            "severity": "notice",
            "title": "app not responding",
            "subtitle": ", ".join(stalled[:3]),
            "rows": [("count", str(len(stalled)))],
            "hazard": "",
        })

    return out


def _check_fleet(now: float) -> list[dict]:
    out = []
    for device in db.list_devices():
        if device["status"] != "registered":
            continue
        silent_for = now - device["last_seen"]
        # Past ALERT_DEVICE_OFFLINE_MAX_SECONDS this is presumed retired/
        # unplugged for good rather than a live problem -- see that
        # constant's comment. Without this a screen taken out of service
        # nags the rest of the fleet's alert feed forever.
        if config.ALERT_DEVICE_OFFLINE_SECONDS < silent_for <= config.ALERT_DEVICE_OFFLINE_MAX_SECONDS:
            hours = silent_for / 3600
            ago = f"{hours:.0f}h ago" if hours >= 1 else f"{silent_for / 60:.0f}m ago"
            out.append({
                "id": f"fleet:offline:{device['device_id']}",
                "severity": "notice",
                "title": "screen offline",
                "subtitle": device.get("label") or device["device_id"],
                "rows": [("last seen", ago)],
                "hazard": "",
            })
    return out


def _check_local_automation(now: float) -> list[dict]:
    """One-shot alert files a separate personal automation suite
    (~/LocalAutomation, not part of this repo) writes to a shared queue —
    weekend art-opening notices (taos_event_scraper.py), network_monitor.py,
    and web_health_monitor.py all write here via that project's own
    common.build_alert_envelope/write_alert_file, explicitly "for the
    local screen net app" per its own code comments. Read-only: this hub
    never deletes/archives these files (same freshly-recomputed-every-cycle
    model as every other check here) — LOCAL_AUTOMATION_ALERT_MAX_AGE_SECONDS
    is the only thing that keeps an old file from showing forever."""
    out = []
    alerts_dir = config.LOCAL_AUTOMATION_DIR / "data" / "alerts_queue"
    if not alerts_dir.is_dir():
        return out

    for path in alerts_dir.glob("*.json"):
        try:
            payload = json.loads(path.read_text(encoding="utf-8"))
            ts = datetime.fromisoformat(payload["timestamp"]).timestamp()
        except Exception as e:
            logger.warning("alerts: skipping unreadable local-automation file %s: %s", path.name, e)
            continue
        if now - ts > config.LOCAL_AUTOMATION_ALERT_MAX_AGE_SECONDS:
            continue

        details = payload.get("details") or {}
        rows = [("summary", payload.get("summary", ""))]
        rows += [(k, str(v)) for k, v in details.items() if v][:3]

        out.append({
            "id": f"local-automation:{path.stem}",
            "severity": "notice",  # this hub's own tier -- never a takeover
            "title": payload.get("title", "local automation"),
            "subtitle": payload.get("source", ""),
            "rows": rows,
            # Only the producer's own "critical" rates the red banner --
            # "info"/"warning" (most of these) render as a plain card.
            "hazard": payload.get("summary", "") if payload.get("severity") == "critical" else "",
        })
    return out


# ---------------------------------------------------------------------------
# State (which urgent alerts already got their takeover)
# ---------------------------------------------------------------------------

def _load_state() -> dict:
    if STATE_PATH.exists():
        try:
            return json.loads(STATE_PATH.read_text())
        except Exception:
            pass
    return {"taken_over": []}


def _save_state(state: dict):
    try:
        STATE_PATH.write_text(json.dumps(state))
    except Exception as e:
        logger.warning("alerts: failed to persist state: %s", e)