Northern-Thailand-Ping-River-Monitor/scripts/e2e_notify.py

"""Drive the production notify path in-process: startup init -> seeded readings
-> forecast cache -> _notify_transitions -> sqlite state -> real ntfy."""

import asyncio
import datetime
import json
import os
import sys

import requests

os.environ.update(
    DB_TYPE="sqlite",
    WATER_DB_PATH=os.path.join(os.environ["LOCALAPPDATA"], "Temp", "smoke3.db"),
    NTFY_SERVER="http://127.0.0.1:2586",
    NTFY_TOKEN=os.environ.get("NTFY_TOKEN", ""),
    NTFY_TOPIC_PREFIX="ping",
)
for f in ("smoke3.db",):
    p = os.path.join(os.environ["LOCALAPPDATA"], "Temp", f)
    if os.path.exists(p):
        os.remove(p)

from src import web_api  # noqa: E402
from src.config import Config  # noqa: E402

assert Config.NTFY_SERVER


async def main():
    # what the lifespan does at startup, minus the scheduler
    from src import notify as notify_mod
    from src.forecast_history import ForecastHistoryStore
    from src.water_scraper_v3 import EnhancedWaterMonitorScraper

    db_config = Config.get_database_config()
    web_api.app_state["scraper"] = EnhancedWaterMonitorScraper(db_config)
    store = ForecastHistoryStore(db_config["connection_string"], db_config["type"])
    store.connect()
    web_api.app_state["forecast_store"] = store
    state = notify_mod.NotificationState(store.engine, store.db_type)
    pub = notify_mod.NtfyPublisher(
        Config.NTFY_SERVER, prefix=Config.NTFY_TOPIC_PREFIX, token=Config.NTFY_TOKEN
    )
    web_api.app_state["notify"] = (pub, state)

    scraper = web_api.app_state["scraper"]
    now = datetime.datetime.now().replace(minute=0, second=0, microsecond=0)

    def seed(level_p1, level_p103, ts):
        rows = [
            {
                "station_code": "P.1",
                "station_id": 1,
                "timestamp": ts,
                "water_level": level_p1,
                "discharge": 400.0,
                "station_name_en": "Nawarat Bridge",
                "station_name_th": "สะพานนวรัฐ",
                "discharge_percent": 30.0,
                "status": "active",
            },
            {
                "station_code": "P.103",
                "station_id": 2,
                "timestamp": ts,
                "water_level": level_p103,
                "discharge": 300.0,
                "station_name_en": "Ring Road 3",
                "station_name_th": "วงแหวน 3",
                "discharge_percent": 20.0,
                "status": "active",
            },
        ]
        scraper.db_adapter.save_measurements(rows)

    def forecast(p):
        with web_api.FORECAST_CACHE_LOCK:
            web_api.FORECAST_CACHE["all"] = (
                0,
                [
                    {
                        "station_code": "P.1",
                        "horizon_hours": 24,
                        "p_warning": p,
                        "predicted_max_level": 3.9,
                        "source": "model",
                    }
                ],
            )

    def poll(topic):
        out = []
        for line in (
            requests.get(f"{Config.NTFY_SERVER}/{topic}/json?poll=1", timeout=5)
            .text.strip()
            .splitlines()
        ):
            m = json.loads(line)
            if m.get("event") == "message":
                out.append(m.get("title") or m.get("message", "")[:40])
        return out

    # cycle 1: quiet
    seed(1.6, 3.2, now - datetime.timedelta(hours=2))
    forecast(0.02)
    await web_api._notify_transitions()
    # cycle 2: P.1 crosses warning, model outlook on
    seed(3.75, 3.3, now - datetime.timedelta(hours=1))
    forecast(0.7)
    await web_api._notify_transitions()
    # cycle 3: same state -> silence
    seed(3.80, 3.3, now)
    forecast(0.65)
    await web_api._notify_transitions()

    print("ping-p1-warning:", poll("ping-p1-warning"))
    print("ping-warning:   ", poll("ping-warning"))
    print("ping-p1-outlook:", poll("ping-p1-outlook"))
    print("ping-p103-warning:", poll("ping-p103-warning"))
    from sqlalchemy import text

    with store.engine.connect() as c:
        print(
            "state table:",
            c.execute(
                text("SELECT key, state, value FROM notification_state ORDER BY key")
            ).fetchall(),
        )


asyncio.run(main())