#!/usr/bin/env python3
"""uplink-watchdog — tell me when a LoRaWAN device stops sending.

ChirpStack and The Things Stack will happily let a sensor go quiet for a month
without saying a word. This is a single file, with no dependencies outside the
Python standard library, that notices and tells you.

It listens for the uplink webhooks your network server can already send, and
alerts when a device that was reporting reliably stops. Devices register
themselves from the first uplink; there is nothing to configure per device.

    $ python3 uplink_watchdog.py --port 8000 --ntfy my-secret-topic

Then point your network server's HTTP integration at http://<host>:8000/uplink

    ChirpStack        Applications > your app > Integrations > HTTP
    The Things Stack  Integrations > Webhooks > Add webhook > Custom

MIT licensed. Written by Geosensor (https://geosensor.tech), who run LoRaWAN
networks for a living and got tired of hearing about dead sensors from clients.
"""
import argparse
import json
import os
import smtplib
import ssl
import sys
import threading
import time
import urllib.request
from email.message import EmailMessage
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

__version__ = "1.0.0"

# How many times a device's own average interval may elapse before we call it
# silent. Three is forgiving enough to survive one dropped uplink, tight enough
# to catch a real failure the same day.
TOLERANCE = 3.0
MIN_SILENCE_SECONDS = 1800
EWMA_ALPHA = 0.25

STATE = {}          # dev_key -> {name, last_seen, interval, uplinks, silent}
STATE_LOCK = threading.Lock()
CONFIG = {}


# --------------------------------------------------------------- payloads

def normalise(payload):
    """Reduce a ChirpStack, Things Stack or generic uplink to (key, name)."""
    if not isinstance(payload, dict):
        return None
    # Joins and acks are not evidence that a device is alive and well.
    if payload.get("event") in ("join", "ack", "txack", "status", "error", "location"):
        return None

    info = payload.get("deviceInfo")
    if isinstance(info, dict) and info.get("devEui"):
        return info["devEui"].lower(), info.get("deviceName") or info["devEui"]

    ids = payload.get("end_device_ids")
    if isinstance(ids, dict) and (ids.get("device_id") or ids.get("dev_eui")):
        key = (ids.get("device_id") or ids.get("dev_eui")).lower()
        return key, ids.get("device_id") or key

    for field in ("dev_eui", "devEui", "devEUI", "device_id", "deviceId", "DevEUI"):
        val = payload.get(field)
        if isinstance(val, str) and val.strip():
            return val.strip().lower(), payload.get("deviceName") or val.strip()
    return None


# --------------------------------------------------------------- alerting

def notify(title, body):
    """Best effort on every configured channel; one failure must not stop the rest."""
    if CONFIG.get("ntfy"):
        try:
            urllib.request.urlopen(urllib.request.Request(
                f"https://ntfy.sh/{CONFIG['ntfy']}", data=body.encode(),
                headers={"Title": title, "Priority": "high", "Tags": "warning"}),
                timeout=10)
        except Exception as exc:                              # noqa: BLE001
            print(f"  ntfy failed: {exc}", file=sys.stderr)

    if CONFIG.get("webhook"):
        try:
            urllib.request.urlopen(urllib.request.Request(
                CONFIG["webhook"],
                data=json.dumps({"title": title, "text": body}).encode(),
                headers={"Content-Type": "application/json"}), timeout=10)
        except Exception as exc:                              # noqa: BLE001
            print(f"  webhook failed: {exc}", file=sys.stderr)

    if CONFIG.get("email_to"):
        msg = EmailMessage()
        msg["From"] = CONFIG["smtp_user"]
        msg["To"] = CONFIG["email_to"]
        msg["Subject"] = title
        msg.set_content(body)
        try:
            ctx = ssl.create_default_context()
            port = int(CONFIG["smtp_port"])
            # Many hosts (Hetzner among them) block 465 outbound but allow 587.
            if port == 465:
                with smtplib.SMTP_SSL(CONFIG["smtp_host"], port, context=ctx, timeout=15) as s:
                    s.login(CONFIG["smtp_user"], CONFIG["smtp_pass"])
                    s.send_message(msg)
            else:
                with smtplib.SMTP(CONFIG["smtp_host"], port, timeout=15) as s:
                    s.starttls(context=ctx)
                    s.login(CONFIG["smtp_user"], CONFIG["smtp_pass"])
                    s.send_message(msg)
        except Exception as exc:                              # noqa: BLE001
            print(f"  email failed: {exc}", file=sys.stderr)

    print(f"ALERT: {title}")


def human(seconds):
    seconds = int(seconds)
    if seconds < 3600:
        return f"{seconds // 60} min"
    if seconds < 86400:
        return f"{seconds // 3600} h {(seconds % 3600) // 60} min"
    return f"{seconds // 86400} d {(seconds % 86400) // 3600} h"


def threshold(entry):
    if CONFIG.get("silence"):
        return CONFIG["silence"] * 60
    if entry["interval"]:
        return max(MIN_SILENCE_SECONDS, entry["interval"] * TOLERANCE)
    return MIN_SILENCE_SECONDS


# --------------------------------------------------------------- state

def save_state():
    path = CONFIG.get("state")
    if not path:
        return
    try:
        tmp = path + ".tmp"
        with open(tmp, "w") as fh:
            json.dump(STATE, fh)
        os.replace(tmp, path)           # atomic, so a crash can't truncate state
    except OSError as exc:
        print(f"  could not save state: {exc}", file=sys.stderr)


def load_state():
    path = CONFIG.get("state")
    if path and os.path.exists(path):
        try:
            with open(path) as fh:
                STATE.update(json.load(fh))
            print(f"restored {len(STATE)} device(s) from {path}")
        except (OSError, ValueError) as exc:
            print(f"  could not read state: {exc}", file=sys.stderr)


def record(key, name, now=None):
    now = now or time.time()
    with STATE_LOCK:
        entry = STATE.get(key)
        if entry is None:
            STATE[key] = {"name": name, "last_seen": now, "interval": None,
                          "uplinks": 1, "silent": False}
            print(f"new device: {name}")
            return
        gap = now - entry["last_seen"]
        # An outage-sized gap would poison the learned cadence, so skip it.
        if 0 < gap < 172800:
            entry["interval"] = gap if entry["interval"] is None else (
                EWMA_ALPHA * gap + (1 - EWMA_ALPHA) * entry["interval"])
        entry["last_seen"] = now
        entry["uplinks"] += 1
        entry["name"] = name or entry["name"]
        if entry["silent"]:
            entry["silent"] = False
            notify(f"Recovered: {entry['name']}",
                   f"{entry['name']} is sending again after {human(gap)} of silence.")


def sweep(now=None):
    """Alert on every device that has gone quiet. Returns how many fired."""
    now = now or time.time()
    fired = 0
    with STATE_LOCK:
        items = list(STATE.items())
    for key, entry in items:
        if entry["silent"]:
            continue
        age = now - entry["last_seen"]
        limit = threshold(entry)
        if age > limit:
            with STATE_LOCK:
                entry["silent"] = True
            notify(f"Device silent: {entry['name']}",
                   f"{entry['name']} ({key}) has not sent an uplink for {human(age)}.\n"
                   f"It normally reports about every {human(entry['interval'] or limit)}.")
            fired += 1
    if fired:
        save_state()
    return fired


def sweep_loop():
    while True:
        time.sleep(CONFIG.get("interval", 300))
        try:
            sweep()
            save_state()
        except Exception as exc:                              # noqa: BLE001
            print(f"sweep error: {exc}", file=sys.stderr)


# --------------------------------------------------------------- server

class Handler(BaseHTTPRequestHandler):
    server_version = f"uplink-watchdog/{__version__}"

    def _reply(self, code, body=b""):
        self.send_response(code)
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        if body:
            self.wfile.write(body)

    def do_POST(self):
        if CONFIG.get("path") and self.path.split("?")[0] != CONFIG["path"]:
            return self._reply(404)
        try:
            length = int(self.headers.get("Content-Length") or 0)
        except ValueError:
            return self._reply(400)
        if length <= 0 or length > 262144:
            return self._reply(400)
        try:
            payload = json.loads(self.rfile.read(length))
        except (ValueError, OSError):
            return self._reply(400)

        found = normalise(payload)
        if found:
            record(*found)
        # Non-uplink events are accepted quietly so the network server does not
        # start reporting delivery failures at you.
        self._reply(204)

    def do_GET(self):
        if self.path.split("?")[0] not in ("/", "/status"):
            return self._reply(404)
        with STATE_LOCK:
            now = time.time()
            report = {k: {"name": v["name"], "silent": v["silent"],
                          "uplinks": v["uplinks"],
                          "seconds_since_uplink": int(now - v["last_seen"])}
                      for k, v in STATE.items()}
        self._reply(200, json.dumps(report, indent=2).encode())

    def log_message(self, fmt, *args):
        pass                                    # the interesting events log themselves


# --------------------------------------------------------------- entry point

def selftest():
    """Prove the detection logic without a network server or a network."""
    CONFIG.update({"interval": 300, "state": None, "silence": None})
    STATE.clear()
    ok = True

    def expect(label, cond):
        nonlocal ok
        print(("  PASS  " if cond else "  FAIL  ") + label)
        ok = ok and cond

    expect("chirpstack payload parsed", normalise(
        {"deviceInfo": {"devEui": "AA11", "deviceName": "sensor-1"}}) == ("aa11", "sensor-1"))
    expect("things stack payload parsed", normalise(
        {"end_device_ids": {"device_id": "sensor-2", "dev_eui": "BB22"}}) == ("sensor-2", "sensor-2"))
    expect("join event ignored", normalise({"event": "join", "deviceInfo": {"devEui": "AA11"}}) is None)
    expect("junk ignored", normalise({"nothing": "useful"}) is None)

    base = time.time() - 7200
    for i in range(6):                          # six uplinks, 15 minutes apart
        record("aa11", "sensor-1", now=base + i * 900)
    expect("cadence learnt (~15 min)", 800 < STATE["aa11"]["interval"] < 1000)
    expect("quiet while healthy", sweep(now=base + 5 * 900 + 60) == 0)

    alerts = []
    global notify
    real_notify, notify = notify, lambda t, b: alerts.append(t)
    try:
        expect("alerts once when silent", sweep(now=base + 5 * 900 + 4000) == 1)
        expect("does not repeat", sweep(now=base + 5 * 900 + 9000) == 0)
        record("aa11", "sensor-1", now=base + 5 * 900 + 9100)
        expect("recovery announced", any("Recovered" in a for a in alerts))
    finally:
        notify = real_notify
    print("\nself-test: " + ("PASSED" if ok else "FAILED"))
    return 0 if ok else 1


def main():
    ap = argparse.ArgumentParser(
        description="Alert when a LoRaWAN device stops sending uplinks.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="Point your network server's HTTP integration at http://<host>:<port>/uplink")
    ap.add_argument("--port", type=int, default=8000)
    ap.add_argument("--bind", default="0.0.0.0")
    ap.add_argument("--path", default="/uplink",
                    help="URL path to accept uplinks on (default: /uplink)")
    ap.add_argument("--silence", type=int, metavar="MIN",
                    help="fixed silence window in minutes; default learns each device's cadence")
    ap.add_argument("--interval", type=int, default=300, help="seconds between sweeps")
    ap.add_argument("--state", default="watchdog-state.json",
                    help="file to survive restarts in; empty string to disable")
    ap.add_argument("--ntfy", metavar="TOPIC", help="ntfy.sh topic for phone push")
    ap.add_argument("--webhook", metavar="URL", help="POST alerts as JSON to this URL")
    ap.add_argument("--email-to", metavar="ADDR")
    ap.add_argument("--smtp-host", default=os.environ.get("SMTP_HOST", ""))
    ap.add_argument("--smtp-port", default=os.environ.get("SMTP_PORT", "587"))
    ap.add_argument("--smtp-user", default=os.environ.get("SMTP_USER", ""))
    ap.add_argument("--smtp-pass", default=os.environ.get("SMTP_PASS", ""))
    ap.add_argument("--selftest", action="store_true", help="run built-in tests and exit")
    ap.add_argument("--version", action="version", version=__version__)
    args = ap.parse_args()

    CONFIG.update(vars(args))
    if args.selftest:
        return selftest()

    if not any((args.ntfy, args.webhook, args.email_to)):
        print("Warning: no alert channel given (--ntfy, --webhook or --email-to).\n"
              "         Running anyway; alerts will only be printed here.\n", file=sys.stderr)

    load_state()
    threading.Thread(target=sweep_loop, daemon=True).start()
    server = ThreadingHTTPServer((args.bind, args.port), Handler)
    print(f"uplink-watchdog {__version__} listening on "
          f"http://{args.bind}:{args.port}{args.path}")
    print("Point your network server's HTTP integration here. Ctrl-C to stop.")
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\nstopping")
        save_state()
    return 0


if __name__ == "__main__":
    sys.exit(main())