yaris-obd/modules/app_network.py

"""Phone-hotspot configuration and read-only uplink status."""
import json
import os
import subprocess
import sys


BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
INSTALLED_HELPER = "/usr/local/bin/yaris-uplink"
REPO_HELPER = os.path.join(BASE_DIR, "deploy", "yaris_uplink.py")
PHONE_CONNECTION = "phone-hotspot"
# 2026-08-14 (FEAT-14): dual T3U — wlan1 (T3U #1) hosts the in-car AP; the
# second T3U (wlan2) carries the internet uplink; onboard wlan0 is a fallback.
PRIMARY_INTERFACE = "wlan2"
FALLBACK_INTERFACE = "wlan0"
AP_INTERFACE = "wlan1"
STATUS_PATH = "/var/lib/yaris-obd/uplink-status.json"


class NetworkConfigurationError(RuntimeError):
    pass


def _interface_exists(iface):
    return os.path.exists(f"/sys/class/net/{iface}")


def _resolve_interface():
    if _interface_exists(PRIMARY_INTERFACE):
        return PRIMARY_INTERFACE
    if _interface_exists(FALLBACK_INTERFACE):
        return FALLBACK_INTERFACE
    return PRIMARY_INTERFACE


def _helper_command(action):
    if os.path.isfile(INSTALLED_HELPER):
        return ["sudo", "-n", INSTALLED_HELPER, action]
    return ["sudo", "-n", sys.executable, REPO_HELPER, action]


def _run_helper(action, payload=None):
    process = subprocess.run(
        _helper_command(action),
        input=json.dumps(payload) if payload is not None else None,
        capture_output=True,
        text=True,
        timeout=120,
    )
    if process.returncode != 0:
        detail = (process.stderr or process.stdout).strip()
        raise NetworkConfigurationError(
            detail or f"Network helper exited {process.returncode}."
        )


def configure_phone_hotspot(ssid, password):
    _run_helper("configure", {"ssid": ssid, "password": password})


def request_phone_connection():
    process = subprocess.run(
        [
            "sudo",
            "-n",
            "systemctl",
            "start",
            "--no-block",
            "yaris-uplink-wait.service",
        ],
        capture_output=True,
        text=True,
        timeout=5,
    )
    if process.returncode != 0:
        detail = (process.stderr or process.stdout).strip()
        raise NetworkConfigurationError(
            detail or "Could not request a phone-hotspot connection."
        )


def remove_phone_hotspot():
    _run_helper("remove")


def _nmcli(*args):
    process = subprocess.run(
        ["nmcli", *args],
        capture_output=True,
        text=True,
        timeout=5,
    )
    if process.returncode != 0:
        raise NetworkConfigurationError(
            (process.stderr or process.stdout).strip() or "nmcli failed."
        )
    return process.stdout.strip()


def _phone_ssid():
    process = subprocess.run(
        [
            "nmcli",
            "-g",
            "802-11-wireless.ssid",
            "connection",
            "show",
            PHONE_CONNECTION,
        ],
        capture_output=True,
        text=True,
        timeout=5,
    )
    return process.stdout.strip() if process.returncode == 0 else ""


def get_status():
    interface = _resolve_interface()
    values = _nmcli(
        "-g",
        "GENERAL.CONNECTION,GENERAL.STATE,GENERAL.IP4-CONNECTIVITY,IP4.ADDRESS",
        "device",
        "show",
        interface,
    ).splitlines()
    values += [""] * (4 - len(values))
    last_transition = {}
    try:
        with open(STATUS_PATH, encoding="utf-8") as handle:
            last_transition = json.load(handle)
    except FileNotFoundError:
        pass
    except (OSError, json.JSONDecodeError) as error:
        last_transition = {"error": f"Could not read uplink status: {error}"}

    active = values[0]
    connectivity = values[2]
    phone_ssid = _phone_ssid()
    internet_reachable = connectivity == "4 (full)"

    # Check AP state on wlan1 (T3U)
    ap_active = False
    ap_ip = ""
    try:
        ap_active = subprocess.run(
            ["nmcli", "-g", "GENERAL.CONNECTION", "device", "show", AP_INTERFACE],
            capture_output=True, text=True, timeout=5,
        ).stdout.strip() == "Hotspot-local"
        if ap_active:
            ap_ip = subprocess.run(
                ["nmcli", "-g", "IP4.ADDRESS", "device", "show", AP_INTERFACE],
                capture_output=True, text=True, timeout=5,
            ).stdout.strip().split("/", 1)[0]
    except (OSError, subprocess.SubprocessError):
        pass

    # Mode reflects the uplink interface state, not the AP
    uplink_mode = (
        "phone" if active == PHONE_CONNECTION
        else "wifi" if active
        else "disconnected"
    )

    return {
        "active_connection": active,
        "mode": uplink_mode,
        "ap_active": ap_active,
        "ap_ip": ap_ip,
        "state": values[1],
        "internet_reachable": internet_reachable,
        "connectivity": connectivity,
        "interface": interface,
        "ip_address": values[3].split("/", 1)[0],
        "phone_configured": bool(phone_ssid),
        "phone_ssid": phone_ssid,
        "large_upload_allowed": (
            internet_reachable and active != "" and active != "Hotspot"
        ),
        "last_transition": last_transition,
    }