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["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, }