yaris-obd/modules/app_alerts.py

import json
import sqlite3
import threading
import time
from urllib.parse import urlparse

import requests

from modules import (
    app_cellular,
    app_dash,
    app_gps,
    app_health,
    app_obd,
    app_sql,
    app_video,
)


CHECK_INTERVAL = 60
RETRY_INTERVAL = 5 * 60
DASHCAM_CHECK_INTERVAL = 5 * 60
POWER_EVENT_GRACE_SECONDS = 5 * 60

SETTING_DEFAULTS = {
    "alerts_enabled": "0",
    "alert_ntfy_url": "",
    "alert_ntfy_token": "",
    "alert_temperature_c": "80",
    "alert_disk_percent": "90",
    "alert_tracking_backlog": "1000",
    "alert_video_backlog": "10",
    "alert_gps_loss_seconds": "60",
    "alert_repeat_seconds": "3600",
}

ALERT_TITLES = {
    "mil": "Check-engine light",
    "disk": "Pi storage nearly full",
    "upload_backlog": "Upload backlog growing",
    "temperature": "Pi temperature high",
    "power": "Pi power or throttling warning",
    "gps_loss": "GPS fix lost while moving",
}

_thread = None
_last_run_at = None
_last_run_error = None
_last_conditions = {}
_moving_started_at = None
_last_dashcam_check_at = None
_last_dashcam_usage = {"status": "not_checked"}


def _number_setting(key, minimum):
    raw = app_sql.get_setting(key, SETTING_DEFAULTS[key])
    try:
        value = float(raw)
    except (TypeError, ValueError):
        return float(SETTING_DEFAULTS[key]), f"{key} must be numeric"
    if value < minimum:
        return float(SETTING_DEFAULTS[key]), f"{key} must be at least {minimum}"
    return value, None


def get_config():
    errors = {}
    values = {}
    minimums = {
        "alert_temperature_c": 1,
        "alert_disk_percent": 1,
        "alert_tracking_backlog": 1,
        "alert_video_backlog": 1,
        "alert_gps_loss_seconds": 1,
        "alert_repeat_seconds": 60,
    }
    for key, minimum in minimums.items():
        values[key], error = _number_setting(key, minimum)
        if error:
            errors[key] = error

    url = app_sql.get_setting("alert_ntfy_url", "")
    parsed = urlparse(url)
    if url and (parsed.scheme not in {"http", "https"} or not parsed.netloc):
        errors["alert_ntfy_url"] = "alert_ntfy_url must be an http(s) URL"

    values.update({
        "enabled": app_sql.get_setting("alerts_enabled", "0") == "1",
        "ntfy_url": url,
        "ntfy_token": app_sql.get_setting("alert_ntfy_token", ""),
        "errors": errors,
    })
    return values


def send_ntfy(url, token, title, message, priority="default", tags=None):
    headers = {
        "Title": title,
        "Priority": priority,
    }
    if tags:
        headers["Tags"] = ",".join(tags)
    if token:
        headers["Authorization"] = f"Bearer {token}"
    body = message.encode("utf-8")
    if app_cellular.is_configured():
        cellular_failure = None
        try:
            status_code, _response_body = app_cellular.http_post(
                url,
                body,
                headers,
                content_type="text/plain; charset=utf-8",
            )
            if 200 <= status_code < 300:
                return
            cellular_failure = f"HTTP {status_code}"
        except app_cellular.CellularError:
            cellular_failure = "modem error"
        print(
            f"Cellular notification failed ({cellular_failure}); "
            "trying WiFi/Ethernet."
        )

    response = requests.post(
        url,
        data=body,
        headers=headers,
        timeout=5,
    )
    response.raise_for_status()


def _transport_error(error):
    if isinstance(error, app_cellular.CellularError):
        return "Cellular notification failed"
    if isinstance(error, requests.Timeout):
        return "Notification request timed out"
    if isinstance(error, requests.HTTPError):
        status_code = (
            error.response.status_code
            if error.response is not None
            else "unknown"
        )
        return f"Notification server returned HTTP {status_code}"
    if isinstance(error, requests.ConnectionError):
        return "Notification connection failed"
    return "Notification request failed"


def gps_loss_while_moving(speed, fix_age, threshold, now=None):
    global _moving_started_at
    now = time.monotonic() if now is None else now
    if speed is None or speed <= 1:
        _moving_started_at = None
        return False
    if _moving_started_at is None:
        _moving_started_at = now
    last_fix_at = None if fix_age is None else now - fix_age
    loss_started_at = (
        _moving_started_at
        if last_fix_at is None
        else max(_moving_started_at, last_fix_at)
    )
    return now - loss_started_at >= threshold


def _dashcam_usage(now=None):
    global _last_dashcam_check_at, _last_dashcam_usage
    now = time.monotonic() if now is None else now
    if (
        _last_dashcam_check_at is None
        or now - _last_dashcam_check_at >= DASHCAM_CHECK_INTERVAL
    ):
        _last_dashcam_usage = app_dash.dashcam_storage_usage()
        _last_dashcam_check_at = now
    return _last_dashcam_usage


def _power_condition(throttling, now=None):
    now = time.time() if now is None else now
    try:
        event_until = float(
            app_sql.get_setting("alert_power_event_until", "0")
        )
    except (TypeError, ValueError) as error:
        raise ValueError("alert_power_event_until is invalid") from error
    if throttling is None:
        return {
            "active": now < event_until,
            "current": [],
            "occurred": [],
            "newly_occurred": [],
            "recent_event": now < event_until,
            "raw": None,
        }
    current = set(throttling["current"] if throttling else [])
    occurred = set(throttling["occurred"] if throttling else [])
    raw_seen = app_sql.get_setting("alert_throttling_seen")
    if raw_seen is None:
        seen = occurred
        app_sql.set_setting(
            "alert_throttling_seen", json.dumps(sorted(occurred))
        )
    else:
        try:
            seen = set(json.loads(raw_seen))
        except (json.JSONDecodeError, TypeError) as error:
            raise ValueError("alert_throttling_seen is invalid") from error
    newly_occurred = occurred - seen
    if raw_seen is not None and occurred != seen:
        app_sql.set_setting(
            "alert_throttling_seen", json.dumps(sorted(occurred))
        )
    if newly_occurred:
        event_until = now + POWER_EVENT_GRACE_SECONDS
        app_sql.set_setting(
            "alert_power_event_until",
            str(event_until),
        )
    return {
        "active": bool(current) or now < event_until,
        "current": sorted(current),
        "occurred": sorted(occurred),
        "newly_occurred": sorted(newly_occurred),
        "recent_event": now < event_until,
        "raw": throttling["raw"] if throttling else None,
    }


def collect_conditions(config):
    health = app_health.collect_health()
    videos = app_video.list_local_videos()
    pending_videos = sum(1 for video in (videos or {}).get("files", []) if not video["uploaded"])
    pending_tracking = app_sql.count_unsent_tracking_rows()
    speed = app_obd.get_fresh_speed()
    fix_age = app_gps.get_fix_age()

    temperature = health["temperature_c"]
    disk = health["disk"]
    throttling = health["throttling"]
    dashcam_disk = _dashcam_usage()
    power = _power_condition(throttling)
    mil_on = app_obd.get_fresh_mil()
    dtc_codes = [
        item["code"] for item in app_obd.get_fresh_dtc_codes()
    ]
    moving_without_fix = gps_loss_while_moving(
        speed,
        fix_age,
        config["alert_gps_loss_seconds"],
    )

    return {
        "mil": {
            "active": mil_on is True,
            "value": {
                "mil_on": mil_on,
                "dtc_codes": dtc_codes,
            },
            "message": (
                "The vehicle MIL is on."
                + (f" Codes: {', '.join(dtc_codes)}" if dtc_codes else "")
            ),
            "priority": "high",
            "tags": ["warning", "car"],
        },
        "disk": {
            "active": (
                (
                    disk is not None
                    and disk["used_percent"] >= config["alert_disk_percent"]
                )
                or (
                    dashcam_disk["status"] == "ok"
                    and dashcam_disk["used_percent"]
                    >= config["alert_disk_percent"]
                )
            ),
            "value": {
                "pi": disk,
                "dashcam": dashcam_disk,
            },
            "message": (
                (
                    f"Pi storage is {disk['used_percent']:.1f}% full. "
                    if disk
                    else "Pi disk usage is unavailable. "
                )
                + (
                    f"Dashcam storage is {dashcam_disk['used_percent']:.1f}% full."
                    if dashcam_disk["status"] == "ok"
                    else f"Dashcam storage status: {dashcam_disk['status']}."
                )
            ),
            "priority": "high",
            "tags": ["warning", "floppy_disk"],
        },
        "upload_backlog": {
            "active": (
                pending_tracking >= config["alert_tracking_backlog"]
                or pending_videos >= config["alert_video_backlog"]
            ),
            "value": {
                "pending_tracking_rows": pending_tracking,
                "pending_videos": pending_videos,
            },
            "message": (
                f"{pending_tracking} tracking rows and {pending_videos} "
                "dashcam videos are waiting to upload."
            ),
            "priority": "default",
            "tags": ["warning", "outbox_tray"],
        },
        "temperature": {
            "active": (
                temperature is not None
                and temperature >= config["alert_temperature_c"]
            ),
            "value": {"temperature_c": temperature},
            "message": (
                f"Pi CPU temperature is {temperature:.1f} °C."
                if temperature is not None
                else "Pi temperature is unavailable."
            ),
            "priority": "high",
            "tags": ["warning", "thermometer"],
        },
        "power": {
            "active": power["active"],
            "value": power,
            "message": (
                "Pi firmware reports: "
                + ", ".join(power["current"]).replace("_", " ")
                + "."
                if power["current"]
                else (
                    "Pi firmware recorded a recent power/throttling event: "
                    + ", ".join(
                        power["newly_occurred"] or power["occurred"]
                    ).replace("_", " ")
                    + "."
                    if power["recent_event"]
                    else "Pi power and throttling state is normal."
                )
            ),
            "priority": "high",
            "tags": ["warning", "electric_plug"],
        },
        "gps_loss": {
            "active": moving_without_fix,
            "value": {
                "speed_kmh": speed,
                "fix_age_seconds": fix_age,
            },
            "message": (
                f"The vehicle is moving at {speed:.1f} km/h but the GPS fix "
                + (
                    "has never been acquired."
                    if fix_age is None
                    else f"is {fix_age:.0f} seconds old."
                )
                if speed is not None
                else "GPS state unavailable."
            ),
            "priority": "high",
            "tags": ["warning", "round_pushpin"],
        },
    }


def _blank_state():
    return {
        "active": False,
        "active_since": None,
        "last_attempt_at": None,
        "last_sent_at": None,
        "recovery_pending": False,
        "last_recovery_at": None,
        "last_error": None,
        "last_value": None,
    }


def process_conditions(conditions, config, now=None, sender=send_ntfy):
    now = time.time() if now is None else now
    states = app_sql.get_alert_states()
    configured = (
        config["enabled"]
        and bool(config["ntfy_url"])
        and not config["errors"]
    )

    for key, condition in conditions.items():
        previous = states.get(key, _blank_state())
        state = previous.copy()
        was_active = previous["active"]
        state["active"] = condition["active"]
        state["last_value"] = json.dumps(condition["value"], sort_keys=True)

        if condition["active"]:
            if not was_active:
                state["active_since"] = now
                state["recovery_pending"] = False
            due = (
                not was_active
                or state["last_sent_at"] is None
                or now - state["last_sent_at"] >= config["alert_repeat_seconds"]
            )
            retry_ready = (
                not was_active
                or state["last_attempt_at"] is None
                or now - state["last_attempt_at"] >= (
                    CHECK_INTERVAL
                    if key == "power"
                    and condition["value"].get("recent_event")
                    else RETRY_INTERVAL
                )
            )
            if configured and due and retry_ready:
                state["last_attempt_at"] = now
                try:
                    sender(
                        config["ntfy_url"],
                        config["ntfy_token"],
                        ALERT_TITLES[key],
                        condition["message"],
                        condition["priority"],
                        condition["tags"],
                    )
                except (
                    requests.RequestException,
                    app_cellular.CellularError,
                ) as error:
                    state["last_error"] = _transport_error(error)
                else:
                    state["last_sent_at"] = now
                    state["last_error"] = None
        else:
            state["active_since"] = None
            if was_active and state["last_sent_at"] is not None:
                state["recovery_pending"] = True
            retry_ready = (
                was_active
                or state["last_attempt_at"] is None
                or now - state["last_attempt_at"] >= RETRY_INTERVAL
            )
            if configured and state["recovery_pending"] and retry_ready:
                state["last_attempt_at"] = now
                try:
                    sender(
                        config["ntfy_url"],
                        config["ntfy_token"],
                        f"Resolved: {ALERT_TITLES[key]}",
                        f"{ALERT_TITLES[key]} is no longer active.",
                        "default",
                        ["white_check_mark"],
                    )
                except (
                    requests.RequestException,
                    app_cellular.CellularError,
                ) as error:
                    state["last_error"] = _transport_error(error)
                else:
                    state["recovery_pending"] = False
                    state["last_recovery_at"] = now
                    state["last_error"] = None

        app_sql.save_alert_state(key, state)


def check_alerts():
    global _last_run_at, _last_run_error, _last_conditions
    config = get_config()
    conditions = collect_conditions(config)
    process_conditions(conditions, config)
    _last_conditions = conditions
    _last_run_at = time.time()
    _last_run_error = None
    return conditions


def _loop():
    global _last_run_error
    while True:
        try:
            check_alerts()
        except (
            OSError,
            ValueError,
            RuntimeError,
            sqlite3.Error,
            requests.RequestException,
            app_cellular.CellularError,
        ) as error:
            _last_run_error = str(error)
            print("Alert check failed:", error)
        time.sleep(CHECK_INTERVAL)


def start_alerting():
    global _thread
    if _thread is None or not _thread.is_alive():
        _thread = threading.Thread(target=_loop, name="alerts", daemon=True)
        _thread.start()
        print(f"Alert thread started, alive: {_thread.is_alive()}")
    return _thread


def get_status():
    config = get_config()
    states = app_sql.get_alert_states()
    thread_alive = _thread is not None and _thread.is_alive()
    return {
        "enabled": config["enabled"],
        "configured": bool(config["ntfy_url"]) and not config["errors"],
        "config_errors": config["errors"],
        "thread_alive": thread_alive,
        "last_run_at": _last_run_at,
        "last_run_error": (
            _last_run_error
            or ("Alert thread is not running." if not thread_alive else None)
        ),
        "conditions": _last_conditions,
        "states": states,
    }