yaris-obd/modules/app_status.py

import os
import subprocess
import time
from urllib.parse import urlparse

import requests

from modules import (
    app_cellular,
    app_gps,
    app_health,
    app_network,
    app_obd,
    app_remote_config,
    app_upload,
)


SERVICE_UNITS = (
    ("Web app", "yaris-obd.service"),
    ("NetworkManager", "NetworkManager.service"),
    ("Hotspot watcher", "yaris-netwatch.timer"),
    ("Phone uplink wait", "yaris-uplink-wait.service"),
)


def parse_systemctl_show(text):
    values = {}
    for line in text.splitlines():
        key, separator, value = line.partition("=")
        if separator:
            values[key] = value
    restarts = values.get("NRestarts", "")
    return {
        "load_state": values.get("LoadState", "unknown"),
        "active_state": values.get("ActiveState", "unknown"),
        "sub_state": values.get("SubState", "unknown"),
        "unit_file_state": values.get("UnitFileState", "unknown"),
        "result": values.get("Result", "unknown"),
        "restarts": int(restarts) if restarts.isdigit() else None,
    }


def _systemd_unit_status(label, unit):
    try:
        result = subprocess.run(
            [
                "systemctl",
                "show",
                unit,
                "--property=LoadState,ActiveState,SubState,UnitFileState,Result,NRestarts",
            ],
            capture_output=True,
            text=True,
            timeout=5,
            check=True,
        )
    except (OSError, subprocess.SubprocessError) as error:
        return {
            "label": label,
            "unit": unit,
            "ok": False,
            "error": str(error),
        }
    status = parse_systemctl_show(result.stdout)
    status.update(
        {
            "label": label,
            "unit": unit,
            "ok": True,
        }
    )
    return status


def _server_status():
    credentials = app_remote_config.server_credentials()
    server_url = credentials["server_url"]
    device_name = credentials["device_name"]
    api_key = credentials["api_key"]
    parsed = urlparse(server_url) if server_url else None
    status = {
        "configured": bool(server_url),
        "server_url": server_url,
        "host": parsed.netloc if parsed else "",
        "device_name": device_name,
        "api_key_configured": bool(api_key),
        "reachable": False,
        "authenticated": False,
        "latency_ms": None,
        "health_http_status": None,
        "error": None,
    }
    if not server_url:
        status["error"] = "Server URL is not configured."
        return status

    started = time.monotonic()
    try:
        health_response = requests.get(f"{server_url}/health", timeout=5)
    except requests.RequestException as error:
        status["error"] = f"Could not reach server: {error}"
        return status

    status["latency_ms"] = round((time.monotonic() - started) * 1000)
    status["health_http_status"] = health_response.status_code
    if health_response.ok:
        status["reachable"] = True
    else:
        status["error"] = f"/health returned HTTP {health_response.status_code}."
        return status

    if not device_name or not api_key:
        status["error"] = "Device credentials are not fully configured yet."
        return status

    try:
        auth_response = requests.get(
            f"{server_url}/api/device-config",
            headers={
                "X-Device-Name": device_name,
                "X-Api-Key": api_key,
            },
            timeout=5,
        )
    except requests.RequestException as error:
        status["error"] = f"Server reachable, but auth check failed: {error}"
        return status

    if auth_response.status_code == 200:
        status["authenticated"] = True
        return status
    if auth_response.status_code == 401:
        status["error"] = "Server reachable, but device credentials were rejected."
        return status
    status["error"] = (
        f"Server reachable, but config auth returned HTTP {auth_response.status_code}."
    )
    return status


def _cellular_hat_status():
    serial_port = app_gps.SERIAL_PORT
    serial_port_present = os.path.exists(serial_port)
    module_responding = False
    if serial_port_present:
        with app_gps._serial_lock:
            module_responding = app_gps._module_responds()
    return {
        "serial_port": serial_port,
        "serial_port_present": serial_port_present,
        "module_responding": module_responding,
        "configured": app_cellular.is_configured(),
        "apn_configured": bool(app_cellular.get_apn()),
    }


def _obd_status():
    port = app_obd._find_obd_port()
    connected = app_obd._is_car_connected()
    return {
        "adapter_present": bool(port),
        "port": port,
        "car_connected": connected,
        "status_label": "Connected" if connected else "Disconnected",
    }


def collect_status():
    try:
        uplink = {"ok": True, **app_network.get_status()}
    except app_network.NetworkConfigurationError as error:
        uplink = {"ok": False, "error": str(error)}

    return {
        "generated_at": time.time(),
        "health": app_health.collect_health(),
        "gps": app_gps.get_status(),
        "uplink": uplink,
        "server": _server_status(),
        "upload": app_upload.get_status(),
        "cellular_hat": _cellular_hat_status(),
        "obd": _obd_status(),
        "services": [
            _systemd_unit_status(label, unit) for label, unit in SERVICE_UNITS
        ],
    }