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"
PRIMARY_INTERFACE = "wlan1"
FALLBACK_INTERFACE = "wlan0"
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)"
    return {
        "active_connection": active,
        "mode": (
            "phone"
            if active == PHONE_CONNECTION
            else "local_hotspot"
            if active == "Hotspot"
            else "wifi"
            if active
            else "disconnected"
        ),
        "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,
    }