Function get_network_topology

webapp_modern.py:5929–6204

Build a topology graph with gateway as center, device types classified, and links inferred.

()

# Kiosk install/teardown runs on a background thread and can take many minutes
# on a slow board — the installer apt-installs chromium. The UI polls
# /api/kiosk/status while that happens, so the task's progress and any failure
# have to be visible there. Without it the API reported 'not_installed' for the
# whole install, which is indistinguishable from "nothing happened", and the UI
# gave up after ~25s on a job that was running fine.
_kiosk_task = {'busy': False, 'phase': 'idle', 'error': None, 'started': 0.0}
_kiosk_task_lock = threading.Lock()


def _kiosk_task_set(**kw) -> None:
    with _kiosk_task_lock:
        _kiosk_task.update(kw)


def _kiosk_task_get() -> dict:
    with _kiosk_task_lock:
        return dict(_kiosk_task)


def _kiosk_mode() -> str:
    """Return 'service', 'autostart', or 'not_installed' based on what's on disk."""
    if _kiosk_autostart_paths():
        return 'autostart'
    if os.path.exists(KIOSK_SERVICE_FILE):
        return 'service'
    return 'not_installed'


def _kiosk_running() -> bool:
    """Best-effort: is a kiosk chromium process running right now?"""
    try:
        proc = subprocess.run(
            ['pgrep', '-f', 'ragnar-kiosk-chromium'],
            capture_output=True, text=True, timeout=5, check=False
        )
        return proc.returncode == 0 and bool(proc.stdout.strip())
    except Exception:
        return False


def _x11_display_for(uid: int) -> str:
    """The DISPLAY of this user's X session, via loginctl. '' when there is none.

    Pi OS Desktop is Wayland now, but plenty of boxes still run X11 — the user
    switched it in raspi-config, or the image predates labwc. Those sessions have
    no wayland-* socket at all, which is the only thing the spawn used to look
    for, so enabling the kiosk from the web UI appeared to do nothing until the
    person at the screen logged out and back in.
    """
    try:
        listing = subprocess.run(['loginctl', 'list-sessions', '--no-legend'],
                                 capture_output=True, text=True, timeout=5, check=False)
        for line in listing.stdout.splitlines():
            parts = line.split()
            if not parts:
                continue
            sid = parts[0]
            props = subprocess.run(