wrolpi/controller/lib/admin.py

"""
Admin operations for WROLPi Controller.

Migrated from wrolpi/admin.py - provides system control functions.
"""

import asyncio
import enum
import json
import logging
import subprocess
import time
from pathlib import Path

import yaml

from controller.lib import config as config_lib
from controller.lib import fstab as fstab_lib
from controller.lib import captive_portal as captive_portal_lib
from controller.lib.config import is_docker_mode, get_config_value, get_media_directory

logger = logging.getLogger(__name__)

# Constants - use controller's own config instead of wrolpi module
DEFAULT_CPU_FREQUENCY = get_config_value('throttle.default_governor', 'ondemand')


class HotspotStatus(enum.Enum):
    """Hotspot status enum."""
    disconnected = enum.auto()  # Radio is on, but Hotspot is not connected.
    off = enum.auto()  # Radio is software-disabled. Can be turned on.
    connected = enum.auto()  # Radio is on, Hotspot is on.
    unknown = enum.auto()  # Unknown status. No nmcli, Docker, or no WiFi hardware.
    in_use = enum.auto()  # Wi-Fi device is in use for a network connection.


class BluetoothStatus(enum.Enum):
    """Bluetooth radio status."""
    on = enum.auto()  # Bluetooth radio is on (rfkill unblocked).
    off = enum.auto()  # Bluetooth radio is off (rfkill blocked).
    unavailable = enum.auto()  # No Bluetooth hardware or rfkill not installed.
    unknown = enum.auto()  # Docker mode or cannot determine.


class DesktopStatus(enum.Enum):
    """Graphical desktop (display manager) status."""
    on = enum.auto()  # Display manager is running.
    off = enum.auto()  # Display manager is installed, but stopped.
    unavailable = enum.auto()  # No display manager installed.
    unknown = enum.auto()  # Docker mode or cannot determine.


class VncStatus(enum.Enum):
    """VNC server status."""
    on = enum.auto()  # VNC server is running.
    off = enum.auto()  # A VNC server is installed, but stopped.
    unavailable = enum.auto()  # No usable VNC server installed.
    unknown = enum.auto()  # Docker mode or cannot determine.


class GovernorStatus(enum.Enum):
    """CPU governor status enum matching wrolpi/admin.py"""
    ondemand = enum.auto()
    powersave = enum.auto()
    unknown = enum.auto()


def get_hotspot_device() -> str:
    return get_config_value('hotspot.device', 'wlan0')


def get_hotspot_ssid() -> str:
    return get_config_value('hotspot.ssid', 'WROLPi')


def get_hotspot_password() -> str:
    return get_config_value('hotspot.password', 'wrolpi hotspot')


# WPA2 uses nmcli's default `wpa-psk` key management; WPA3 uses `sae`.
HOTSPOT_PROTOCOLS = ('wpa2', 'wpa3')


def get_hotspot_protocol() -> str:
    return get_config_value('hotspot.protocol', 'wpa2')


def get_device_hotspot_protocols(device: str) -> list[str]:
    """
    Probe a WiFi device with `iw` and report which hotspot protocols it can host.

    WPA2 only requires AP mode.  WPA3 requires SAE in AP mode, which is only possible
    when the firmware offloads SAE for APs (SAE_OFFLOAD_AP), or the driver is
    mac80211-based (advertises AP/VLAN) with the CMAC cipher so wpa_supplicant can run
    SAE in software.  Notably the Raspberry Pi's brcmfmac driver supports neither.
    """
    if is_docker_mode():
        return []

    try:
        # Absolute path like rfkill; `iw` is in /usr/sbin which is not on every PATH.
        result = subprocess.run(
            ["/usr/sbin/iw", "dev", device, "info"],
            capture_output=True,
            text=True,
            timeout=5,
        )
        if result.returncode != 0:
            return []
        wiphy = None
        for line in result.stdout.splitlines():
            stripped = line.strip()
            if stripped.startswith("wiphy "):
                wiphy = stripped.split()[1]
                break
        if wiphy is None:
            return []

        result = subprocess.run(
            ["/usr/sbin/iw", "phy", f"phy{wiphy}", "info"],
            capture_output=True,
            text=True,
            timeout=5,
        )
        if result.returncode != 0:
            return []
    except (FileNotFoundError, subprocess.TimeoutExpired, subprocess.SubprocessError):
        # Cannot probe without `iw`; assume the WPA2 hotspot nmcli has always created.
        return ['wpa2']

    modes, ciphers, ext_features = set(), set(), set()
    section = None
    for line in result.stdout.splitlines():
        stripped = line.strip()
        if stripped.startswith('Supported interface modes:'):
            section = 'modes'
            continue
        elif stripped.startswith('Supported Ciphers:'):
            section = 'ciphers'
            continue
        elif stripped.startswith(('Supported extended features:', 'Extended features:')):
            # Real Pi hardware prints "Supported extended features:"; older iw
            # releases print "Extended features:".
            section = 'ext'
            continue
        if not stripped.startswith('*'):
            section = None
            continue
        value = stripped.lstrip('* ').strip()
        if section == 'modes':
            modes.add(value)
        elif section == 'ciphers':
            ciphers.add(value.split()[0])
        elif section == 'ext' and '[' in stripped and ']' in stripped:
            ext_features.add(stripped.split('[', 1)[1].split(']', 1)[0].strip())

    protocols = []
    if 'AP' in modes:
        protocols.append('wpa2')
        sae_offload_ap = 'SAE_OFFLOAD_AP' in ext_features
        software_sae = 'AP/VLAN' in modes and any('CMAC' in cipher for cipher in ciphers)
        if sae_offload_ap or software_sae:
            protocols.append('wpa3')
    return protocols


def update_hotspot_settings(device: str = None, ssid: str = None, password: str = None,
                            protocol: str = None, captive_portal: bool = None) -> dict:
    """
    Update hotspot settings in controller.yaml.  Only provided values are changed.

    Settings are applied when the hotspot starts, so a change saved while the hotspot is up
    reports `restart_required`.

    Returns:
        dict with success status and the current settings
    """
    if is_docker_mode():
        return {"success": False, "error": "Not available in Docker mode"}

    if device is not None and not device.strip():
        return {"success": False, "error": "Hotspot device must not be empty"}
    if ssid is not None and not ssid.strip():
        return {"success": False, "error": "Hotspot SSID must not be empty"}
    # nmcli requires a WPA password of 8 to 63 characters.
    if password is not None and not 8 <= len(password) <= 63:
        return {"success": False, "error": "Hotspot password must be 8 to 63 characters"}
    if protocol is not None and protocol not in HOTSPOT_PROTOCOLS:
        return {"success": False, "error": f"Hotspot protocol must be one of: {', '.join(HOTSPOT_PROTOCOLS)}"}

    changed = False
    if device:
        config_lib.update_config('hotspot.device', device)
        changed = True
    if ssid:
        config_lib.update_config('hotspot.ssid', ssid)
        changed = True
    if password:
        config_lib.update_config('hotspot.password', password)
        changed = True
    if protocol:
        config_lib.update_config('hotspot.protocol', protocol)
        changed = True
    if captive_portal is not None:
        config_lib.update_config('hotspot.captive_portal', bool(captive_portal))
        changed = True

    if changed:
        try:
            config_lib.save_config()
        except RuntimeError as e:
            return {"success": False, "error": str(e)}

    logger.info("Hotspot settings updated (device=%s, ssid=%s, protocol=%s, captive_portal=%s)",
                get_hotspot_device(), get_hotspot_ssid(), get_hotspot_protocol(),
                captive_portal_lib.is_captive_portal_enabled())
    return {
        "success": True,
        "device": get_hotspot_device(),
        "ssid": get_hotspot_ssid(),
        "password": get_hotspot_password(),
        "protocol": get_hotspot_protocol(),
        "captive_portal": captive_portal_lib.is_captive_portal_enabled(),
        # dnsmasq only reads its include when the Hotspot connection activates.
        "restart_required": changed and get_hotspot_status() == HotspotStatus.connected,
    }


def migrate_hotspot_settings_from_wrolpi_config():
    """
    One-time copy of hotspot settings from wrolpi.yaml into controller.yaml.

    Hotspot settings used to be saved in wrolpi.yaml by the Settings page; the
    Controller now owns them in controller.yaml.  Copy any values the user set
    in wrolpi.yaml unless controller.yaml already sets them explicitly.
    Called on Controller startup after the primary drive is mounted.
    """
    if not config_lib.CONFIG_PATH_ON_DRIVE.exists():
        return

    try:
        with open(config_lib.CONFIG_PATH_ON_DRIVE) as f:
            controller_config = yaml.safe_load(f) or {}
    except (IOError, yaml.YAMLError) as e:
        logger.warning("Failed to read controller.yaml for hotspot migration: %s", e)
        return

    wrolpi_config_path = get_media_directory() / 'config' / 'wrolpi.yaml'
    if not wrolpi_config_path.exists():
        return

    try:
        with open(wrolpi_config_path) as f:
            wrolpi_config = yaml.safe_load(f) or {}
    except (IOError, yaml.YAMLError) as e:
        logger.warning("Failed to read wrolpi.yaml for hotspot migration: %s", e)
        return

    explicit = controller_config.get('hotspot') or {}
    key_map = {'device': 'hotspot_device', 'ssid': 'hotspot_ssid', 'password': 'hotspot_password'}
    changed = False
    for controller_key, wrolpi_key in key_map.items():
        value = wrolpi_config.get(wrolpi_key)
        if value and controller_key not in explicit and value != get_config_value(f'hotspot.{controller_key}'):
            config_lib.update_config(f'hotspot.{controller_key}', value)
            changed = True

    # A user who never wants the hotspot (e.g. to avoid broadcasting) set this false.
    on_startup = wrolpi_config.get('hotspot_on_startup')
    if isinstance(on_startup, bool) and 'on_startup' not in explicit \
            and on_startup != get_config_value('hotspot.on_startup'):
        config_lib.update_config('hotspot.on_startup', on_startup)
        changed = True

    if changed:
        try:
            config_lib.save_config()
            logger.info("Migrated hotspot settings from wrolpi.yaml to controller.yaml")
        except RuntimeError as e:
            logger.warning("Could not save migrated hotspot settings: %s", e)


# In /run so it is cleared by every reboot.
HOTSPOT_BOOT_MARKER = Path("/run/wrolpi-controller/hotspot-on-startup")
HOTSPOT_BOOT_ATTEMPTS = 3
HOTSPOT_BOOT_RETRY_SECONDS = 10


def should_start_hotspot_on_boot() -> bool:
    """True unless the user set `hotspot.on_startup` false.

    The user's choice is in controller.yaml on the primary drive.  When it cannot be read, the hotspot is only
    started on a fresh install (no drive in /etc/fstab), which needs it to reach the onboarding wizard; a configured
    drive which failed to mount must not start a hotspot the user disabled.
    """
    if is_docker_mode():
        return False
    if not config_lib.CONFIG_PATH_ON_DRIVE.exists():
        media_directory = str(get_media_directory())
        try:
            entries = fstab_lib.parse_fstab()
        except OSError as e:
            logger.warning("Cannot read fstab to decide on the hotspot: %s", e)
            return False
        return not any(e.get("mount_point") == media_directory for e in entries)
    return get_config_value("hotspot.on_startup", True) is True


def start_hotspot_on_boot():
    """Start the hotspot once per boot (not when the Controller restarts), unless the user disabled it."""
    if HOTSPOT_BOOT_MARKER.exists():
        return
    try:
        HOTSPOT_BOOT_MARKER.parent.mkdir(parents=True, exist_ok=True)
        HOTSPOT_BOOT_MARKER.touch()
    except OSError as e:
        logger.warning("Cannot write %s, the hotspot may restart with the Controller: %s", HOTSPOT_BOOT_MARKER, e)

    if not should_start_hotspot_on_boot():
        logger.info("Not starting the hotspot on boot")
        return

    # The WiFi device may still be starting early in boot.
    for attempt in range(1, HOTSPOT_BOOT_ATTEMPTS + 1):
        result = start_hotspot()
        if result.get("success"):
            logger.info("Started the hotspot on boot")
            return
        logger.warning("Failed to start the hotspot on boot (attempt %s): %s", attempt, result.get("error"))
        if attempt < HOTSPOT_BOOT_ATTEMPTS:
            time.sleep(HOTSPOT_BOOT_RETRY_SECONDS)


def get_wifi_devices() -> list[str]:
    """
    List WiFi device names (e.g. wlan0, wlp2s0) using nmcli.

    Returns an empty list in Docker mode or when nmcli is unavailable.
    """
    if is_docker_mode():
        return []

    try:
        result = subprocess.run(
            ["nmcli", "-t", "-f", "DEVICE,TYPE", "device"],
            capture_output=True,
            text=True,
            timeout=5,
        )
        if result.returncode != 0:
            return []

        devices = []
        for line in result.stdout.strip().splitlines():
            parts = line.split(":")
            # Exact "wifi" match excludes wifi-p2p devices.
            if len(parts) >= 2 and parts[1] == "wifi":
                devices.append(parts[0])
        return devices
    except (FileNotFoundError, subprocess.TimeoutExpired, subprocess.SubprocessError):
        return []


def get_current_ssid(interface: str = None) -> str | None:
    """
    Returns the name of the SSID that is currently connected.
    Returns None if Hotspot is active, or no Wi-Fi network is being used.
    """
    if interface is None:
        interface = get_hotspot_device()
    try:
        result = subprocess.run(
            ['iwgetid', interface, '--raw'],
            capture_output=True,
            text=True,
            timeout=5,
        )
        if result.returncode == 0:
            ssid = result.stdout.strip()
            return ssid if ssid else None
        elif result.returncode == 255:
            # Hotspot is in use
            return None
        else:
            return None
    except (FileNotFoundError, subprocess.TimeoutExpired, subprocess.SubprocessError):
        return None


def get_hotspot_status() -> HotspotStatus:
    """
    Get current hotspot status matching wrolpi/admin.py logic.

    Returns:
        HotspotStatus enum value
    """
    if is_docker_mode():
        return HotspotStatus.unknown

    device = get_hotspot_device()

    # Check if device is connected to a Wi-Fi network (not hotspot)
    if get_current_ssid(device):
        return HotspotStatus.in_use

    # Check nmcli status
    try:
        result = subprocess.run(
            ["nmcli"],
            capture_output=True,
            text=True,
            timeout=5,
        )
        if result.returncode != 0:
            return HotspotStatus.unknown

        output = result.stdout.strip()
        for line in output.splitlines():
            if line.startswith(f'{device}: connected'):
                return HotspotStatus.connected
            elif line.startswith(f'{device}: disconnected'):
                return HotspotStatus.disconnected
            elif line.startswith(f'{device}: unavailable'):
                return HotspotStatus.off

        return HotspotStatus.unknown

    except subprocess.TimeoutExpired:
        return HotspotStatus.unknown
    except FileNotFoundError:
        return HotspotStatus.unknown
    except subprocess.SubprocessError:
        return HotspotStatus.unknown


def get_hotspot_status_dict() -> dict:
    """
    Get hotspot status as a dict for API responses.

    Returns dict matching HotspotStatusResponse schema:
        enabled: bool - Whether hotspot is currently enabled
        available: bool - Whether hotspot functionality is available
        reason: Optional[str] - Reason if unavailable
        ssid: Optional[str] - Hotspot SSID when enabled
        device: Optional[str] - WiFi device name
        ip: Optional[str] - Hotspot IPv4 address when enabled
        captive_portal: bool - Whether the captive portal is enabled
        portal_url: Optional[str] - Landing page URL when enabled
    """
    device = get_hotspot_device()
    status = get_hotspot_status()

    # Map enum status to enabled/available flags
    enabled = status == HotspotStatus.connected
    # Hotspot is available even when radio is off because start_hotspot()
    # turns the radio on first.
    available = status != HotspotStatus.unknown

    reason = None
    if status == HotspotStatus.unknown:
        reason = "Hotspot not supported or nmcli unavailable"
    elif status == HotspotStatus.in_use:
        reason = "WiFi device is connected to a network"

    ip = captive_portal_lib.get_hotspot_ip(device) if enabled else None
    portal_enabled = captive_portal_lib.is_captive_portal_enabled()
    return {
        "enabled": enabled,
        "available": available,
        "reason": reason,
        "ssid": get_hotspot_ssid() if enabled else None,
        "device": device,
        "ip": ip,
        "captive_portal": portal_enabled,
        "portal_url": f"http://{ip}{captive_portal_lib.PORTAL_PATH}" if ip and portal_enabled else None,
    }


def start_hotspot() -> dict:
    """
    Start the WiFi hotspot (turn the radio on and bring up the Hotspot connection).

    Returns:
        dict with success status and message
    """
    if is_docker_mode():
        return {"success": False, "error": "Not available in Docker mode"}

    device = get_hotspot_device()
    ssid = get_hotspot_ssid()
    password = get_hotspot_password()
    protocol = get_hotspot_protocol()

    try:
        # First ensure radio is on
        subprocess.run(
            ["nmcli", "radio", "wifi", "on"],
            capture_output=True,
            text=True,
            timeout=10,
        )

        # Wait for device to become ready after radio on.
        # On real hardware, the device transitions from 'unavailable' to 'disconnected'
        # which can take a few seconds.
        import time
        device_ready = False
        for _ in range(20):  # 20 * 0.5s = 10s max
            check = subprocess.run(
                ["nmcli", "-t", "-f", "DEVICE,STATE", "device", "status"],
                capture_output=True,
                text=True,
                timeout=5,
            )
            if check.returncode == 0:
                for line in check.stdout.strip().splitlines():
                    parts = line.split(":")
                    if len(parts) >= 2 and parts[0] == device and parts[1] != "unavailable":
                        device_ready = True
                        break
            if device_ready:
                break
            time.sleep(0.5)

        if not device_ready:
            logger.warning("WiFi device %s did not become ready after turning radio on", device)
            return {"success": False, "error": f"WiFi device {device} not ready after turning radio on"}

        # dnsmasq reads the captive portal include only when the connection activates, so it
        # must be in place first.  The address is not known until then; NetworkManager almost
        # always picks the default, and the file is corrected below if it did not.
        captive_portal_lib.apply_dnsmasq_config(captive_portal_lib.dnsmasq_config_ip()
                                                or captive_portal_lib.DEFAULT_HOTSPOT_IP)
        # A new hotspot session: every client sees the portal page again.
        captive_portal_lib.reset_acknowledged_clients()

        # Create hotspot using NetworkManager
        logger.info("Starting hotspot on %s with SSID %s", device, ssid)
        result = subprocess.run(
            [
                "nmcli", "device", "wifi", "hotspot",
                "ifname", device,
                "ssid", ssid,
                "password", password,
            ],
            capture_output=True,
            text=True,
            timeout=30,
        )

        if result.returncode != 0:
            logger.warning("Failed to start hotspot: %s", result.stderr.strip())
            return {"success": False, "error": result.stderr.strip() or "Unknown error"}

        # Explicitly set the key management so a reused "Hotspot" profile does not
        # keep a previously configured protocol.  WPA3 (SAE) requires PMF.
        key_mgmt = "sae" if protocol == "wpa3" else "wpa-psk"
        pmf = "3" if protocol == "wpa3" else "0"  # 3=required, 0=default
        modify = subprocess.run(
            ["nmcli", "connection", "modify", "Hotspot",
             "802-11-wireless-security.key-mgmt", key_mgmt,
             "802-11-wireless-security.pmf", pmf],
            capture_output=True,
            text=True,
            timeout=10,
        )
        up = None
        if modify.returncode == 0:
            # Re-activate so the profile change takes effect.
            up = subprocess.run(
                ["nmcli", "connection", "up", "Hotspot"],
                capture_output=True,
                text=True,
                timeout=30,
            )

        if modify.returncode != 0 or up.returncode != 0:
            error = (modify.stderr if modify.returncode != 0 else up.stderr).strip()
            if protocol == "wpa3":
                # The driver likely rejected SAE; revert so the hotspot still works.
                logger.warning("Failed to start WPA3 hotspot, reverting to WPA2: %s", error)
                revert_modify = subprocess.run(
                    ["nmcli", "connection", "modify", "Hotspot",
                     "802-11-wireless-security.key-mgmt", "wpa-psk",
                     "802-11-wireless-security.pmf", "0"],
                    capture_output=True, text=True, timeout=10,
                )
                revert_up = None
                if revert_modify.returncode == 0:
                    revert_up = subprocess.run(
                        ["nmcli", "connection", "up", "Hotspot"],
                        capture_output=True, text=True, timeout=30,
                    )
                if revert_modify.returncode == 0 and revert_up.returncode == 0:
                    # The hotspot is up (as WPA2), so it still needs its captive portal address.
                    _fix_captive_portal_ip(device)
                    return {"success": False,
                            "error": f"WPA3 is not supported by {device}; hotspot reverted to WPA2: {error}"}
                revert_error = (revert_modify.stderr if revert_modify.returncode != 0 else revert_up.stderr).strip()
                logger.error("Reverting hotspot to WPA2 also failed: %s", revert_error)
                return {"success": False,
                        "error": f"WPA3 failed on {device} and falling back to WPA2 also failed;"
                                 f" the hotspot may be down: {error}"}
            # WPA2 is what nmcli created the hotspot with anyway; not fatal.
            logger.warning("Failed to explicitly set WPA2 on hotspot: %s", error)

        _fix_captive_portal_ip(device)

        logger.info("Hotspot started successfully on %s (%s)", device, protocol)
        return {"success": True, "ssid": ssid, "device": device}

    except subprocess.TimeoutExpired:
        logger.warning("Timeout starting hotspot on %s", device)
        return {"success": False, "error": "Timeout starting hotspot"}
    except FileNotFoundError:
        logger.warning("nmcli not found, cannot start hotspot")
        return {"success": False, "error": "nmcli not found"}
    except subprocess.SubprocessError as e:
        logger.warning("Failed to start hotspot: %s", e)
        return {"success": False, "error": str(e)}


def _fix_captive_portal_ip(device: str):
    """
    Re-point the captive portal include at the address NetworkManager actually assigned.

    Rare: NetworkManager uses 10.42.1.1 (and so on) only when another shared connection
    already holds 10.42.0.1.  Re-activating the connection makes dnsmasq re-read the file.
    """
    if not captive_portal_lib.is_captive_portal_enabled():
        return
    ip = captive_portal_lib.get_hotspot_ip(device)
    if not ip or ip == captive_portal_lib.dnsmasq_config_ip():
        return
    logger.info("Hotspot address is %s, rewriting captive portal config and re-activating", ip)
    if not captive_portal_lib.apply_dnsmasq_config(ip):
        return
    try:
        up = subprocess.run(["nmcli", "connection", "up", "Hotspot"], capture_output=True, text=True, timeout=30)
        if up.returncode != 0:
            logger.warning("Could not re-activate hotspot for captive portal; dnsmasq still serves the old"
                           " address: %s", up.stderr.strip())
    except (subprocess.SubprocessError, FileNotFoundError) as e:
        logger.warning("Could not re-activate hotspot for captive portal: %s", e)


def stop_hotspot() -> dict:
    """
    Stop the WiFi hotspot by turning the WiFi radio off.

    Returns:
        dict with success status
    """
    if is_docker_mode():
        return {"success": False, "error": "Not available in Docker mode"}

    logger.info("Stopping hotspot (turning WiFi radio off)")
    # The session is over; a lease reused later must see the portal page again.
    captive_portal_lib.reset_acknowledged_clients()
    try:
        # Turn off the radio to disconnect hotspot
        result = subprocess.run(
            ["nmcli", "radio", "wifi", "off"],
            capture_output=True,
            text=True,
            timeout=10,
        )

        # Success even if already off
        logger.info("Hotspot stopped successfully")
        return {"success": True}

    except subprocess.TimeoutExpired:
        logger.warning("Timeout stopping hotspot")
        return {"success": False, "error": "Timeout stopping hotspot"}
    except FileNotFoundError:
        logger.warning("nmcli not found, cannot stop hotspot")
        return {"success": False, "error": "nmcli not found"}
    except subprocess.SubprocessError as e:
        logger.warning("Failed to stop hotspot: %s", e)
        return {"success": False, "error": str(e)}


# --- Bluetooth ---


async def get_bluetooth_status() -> BluetoothStatus:
    """
    Get current Bluetooth radio status using rfkill.

    Parses JSON output from `rfkill -J` to determine if Bluetooth is blocked.

    Returns:
        BluetoothStatus enum value
    """
    if is_docker_mode():
        return BluetoothStatus.unknown

    try:
        proc = await asyncio.create_subprocess_exec(
            "/usr/sbin/rfkill", "-J",
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=5)

        if proc.returncode != 0:
            return BluetoothStatus.unknown

        data = json.loads(stdout.decode())
        bt_devices = [d for d in data.get("rfkilldevices", []) if d.get("type") == "bluetooth"]

        if not bt_devices:
            return BluetoothStatus.unavailable

        # If any Bluetooth device is soft or hard blocked, report as off.
        for device in bt_devices:
            if device.get("soft") == "blocked" or device.get("hard") == "blocked":
                return BluetoothStatus.off

        return BluetoothStatus.on

    except (json.JSONDecodeError, KeyError):
        return BluetoothStatus.unknown
    except asyncio.TimeoutError:
        return BluetoothStatus.unknown
    except FileNotFoundError:
        return BluetoothStatus.unavailable
    except OSError:
        return BluetoothStatus.unknown


async def get_bluetooth_status_dict() -> dict:
    """
    Get Bluetooth status as a dict for API responses.

    Returns dict matching BluetoothStatusResponse schema:
        enabled: bool - Whether Bluetooth radio is currently on
        available: bool - Whether Bluetooth functionality is available
        reason: Optional[str] - Reason if unavailable
    """
    status = await get_bluetooth_status()

    enabled = status == BluetoothStatus.on
    available = status not in (BluetoothStatus.unknown, BluetoothStatus.unavailable)

    reason = None
    if status == BluetoothStatus.unknown:
        reason = "Bluetooth not supported or rfkill unavailable"
    elif status == BluetoothStatus.unavailable:
        reason = "No Bluetooth hardware detected"

    return {
        "enabled": enabled,
        "available": available,
        "reason": reason,
    }


async def unblock_bluetooth() -> dict:
    """
    Unblock the Bluetooth radio using rfkill (turns it on).

    Returns:
        dict with success status
    """
    if is_docker_mode():
        return {"success": False, "error": "Not available in Docker mode"}

    logger.info("Unblocking Bluetooth radio")
    try:
        proc = await asyncio.create_subprocess_exec(
            "/usr/sbin/rfkill", "unblock", "bluetooth",
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=10)

        if proc.returncode == 0:
            logger.info("Bluetooth radio unblocked successfully")
            return {"success": True}
        else:
            error_msg = stderr.decode().strip()
            logger.warning("Failed to unblock Bluetooth: %s", error_msg)
            return {"success": False, "error": error_msg or "Unknown error"}

    except asyncio.TimeoutError:
        logger.warning("Timeout unblocking Bluetooth")
        return {"success": False, "error": "Timeout unblocking Bluetooth"}
    except FileNotFoundError:
        logger.warning("rfkill not found, cannot unblock Bluetooth")
        return {"success": False, "error": "rfkill not found"}
    except OSError as e:
        logger.warning("Failed to unblock Bluetooth: %s", e)
        return {"success": False, "error": str(e)}


async def block_bluetooth() -> dict:
    """
    Block the Bluetooth radio using rfkill (turns it off).

    Returns:
        dict with success status
    """
    if is_docker_mode():
        return {"success": False, "error": "Not available in Docker mode"}

    logger.info("Blocking Bluetooth radio")
    try:
        proc = await asyncio.create_subprocess_exec(
            "/usr/sbin/rfkill", "block", "bluetooth",
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=10)

        if proc.returncode == 0:
            logger.info("Bluetooth radio blocked successfully")
            return {"success": True}
        else:
            error_msg = stderr.decode().strip()
            logger.warning("Failed to block Bluetooth: %s", error_msg)
            return {"success": False, "error": error_msg or "Unknown error"}

    except asyncio.TimeoutError:
        logger.warning("Timeout blocking Bluetooth")
        return {"success": False, "error": "Timeout blocking Bluetooth"}
    except FileNotFoundError:
        logger.warning("rfkill not found, cannot block Bluetooth")
        return {"success": False, "error": "rfkill not found"}
    except OSError as e:
        logger.warning("Failed to block Bluetooth: %s", e)
        return {"success": False, "error": str(e)}


# --- systemd units ---

async def _unit_properties(unit: str, *properties: str) -> dict | None:
    """
    Read systemd properties for a unit, e.g. LoadState and ActiveState.

    `systemctl show` exits 0 even for a unit that does not exist (reporting
    LoadState=not-found), so callers must judge existence from LoadState.
    Note ConditionResult is useless here: it reports "no" for any unit that has
    never been started, configured or not.

    Returns None when systemctl itself failed.  Timeouts and OSError propagate so
    callers can tell "no systemd here" from "systemd would not answer".
    """
    proc = await asyncio.create_subprocess_exec(
        "systemctl", "show", unit, f"--property={','.join(properties)}",
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=5)

    if proc.returncode != 0:
        return None

    return dict(
        line.split("=", 1) for line in stdout.decode().strip().splitlines() if "=" in line
    )


async def _units_properties(units: list[str], *properties: str) -> list[dict] | None:
    """
    Read the same systemd properties for several units in one `systemctl show` call.

    Output is one blank-line-separated block per unit, in the order asked for.
    Batching matters because these probes run on every /api/stats request.

    Returns one dict per unit, or None when systemctl itself failed.
    """
    proc = await asyncio.create_subprocess_exec(
        "systemctl", "show", *units, f"--property={','.join(properties)}",
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=5)

    if proc.returncode != 0:
        return None

    blocks = []
    for block in stdout.decode().strip().split("\n\n"):
        blocks.append(dict(
            line.split("=", 1) for line in block.strip().splitlines() if "=" in line
        ))
    # Pad so callers can index by unit even if systemd reported fewer blocks.
    while len(blocks) < len(units):
        blocks.append({})
    return blocks


async def _systemctl_unit(action: str, unit: str, label: str) -> dict:
    """Run `systemctl <start|stop> <unit>` and report the result."""
    if is_docker_mode():
        return {"success": False, "error": "Not available in Docker mode"}

    logger.info("%s %s requested (%s)", label, action, unit)
    try:
        proc = await asyncio.create_subprocess_exec(
            "systemctl", action, unit,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=30)

        if proc.returncode == 0:
            logger.info("%s %s succeeded", label, action)
            return {"success": True}
        else:
            error_msg = stderr.decode().strip()
            logger.warning("Failed to %s %s: %s", action, label.lower(), error_msg)
            return {"success": False, "error": error_msg or "Unknown error"}

    except asyncio.TimeoutError:
        logger.warning("Timeout during %s %s", label.lower(), action)
        return {"success": False, "error": f"Timeout during {label.lower()} {action}"}
    except FileNotFoundError:
        logger.warning("systemctl not found, cannot %s %s", action, label.lower())
        return {"success": False, "error": "systemctl not found"}
    except OSError as e:
        logger.warning("Failed to %s %s: %s", action, label.lower(), e)
        return {"success": False, "error": str(e)}


# --- Desktop ---

# Every display manager (lightdm on RPi OS, gdm3/sddm on Debian) provides this
# systemd alias, so the Controller does not need to know which one is installed.
DISPLAY_MANAGER_UNIT = "display-manager.service"


async def get_desktop_status() -> DesktopStatus:
    """
    Get the graphical desktop status by checking the display manager's systemd unit.

    Returns:
        DesktopStatus enum value
    """
    if is_docker_mode():
        return DesktopStatus.unknown

    try:
        properties = await _unit_properties(DISPLAY_MANAGER_UNIT, "LoadState", "ActiveState")
        if properties is None:
            return DesktopStatus.unknown

        if properties.get("LoadState") != "loaded":
            # "not-found" when no display manager is installed (e.g. RPi OS Lite).
            return DesktopStatus.unavailable

        return DesktopStatus.on if properties.get("ActiveState") == "active" else DesktopStatus.off

    except asyncio.TimeoutError:
        return DesktopStatus.unknown
    except FileNotFoundError:
        return DesktopStatus.unavailable
    except OSError:
        return DesktopStatus.unknown


async def get_desktop_status_dict() -> dict:
    """
    Get desktop status as a dict for API responses.

    Returns dict matching DesktopStatusResponse schema:
        running: bool - Whether the display manager is currently running
        available: bool - Whether a display manager is installed
        reason: Optional[str] - Reason if unavailable
    """
    status = await get_desktop_status()

    running = status == DesktopStatus.on
    available = status in (DesktopStatus.on, DesktopStatus.off)

    reason = None
    if status == DesktopStatus.unknown:
        reason = "Desktop not supported or systemctl unavailable"
    elif status == DesktopStatus.unavailable:
        reason = "No display manager installed"

    return {
        "running": running,
        "available": available,
        "reason": reason,
    }


async def _systemctl_desktop(action: str) -> dict:
    """Run `systemctl <start|stop> display-manager.service` and report the result."""
    return await _systemctl_unit(action, DISPLAY_MANAGER_UNIT, "Desktop")


# Stopping the display manager leaves the monitor black with only a blinking
# cursor, even though the console VT is already in the foreground with a getty
# and a shell on it: the display manager hands the display back without the
# console being repainted.  Bouncing through another VT forces a modeset and
# redraw, and a short message guarantees there is something legible on screen.
CONSOLE_VT = "1"
CONSOLE_GETTY_UNIT = "getty@tty1.service"
CONSOLE_DEVICE = "/dev/tty1"
REPAINT_VT = "2"

CONSOLE_MESSAGE = (
    "\n*** The WROLPi Controller stopped the desktop. ***\n"
    "*** Press Enter for a prompt.  Start the desktop again with: ***\n"
    "***   sudo systemctl start display-manager ***\n\n"
)


async def _console_getty_active() -> bool:
    """Whether a getty is running on the console VT (i.e. switching there gives a terminal)."""
    try:
        proc = await asyncio.create_subprocess_exec(
            "systemctl", "is-active", CONSOLE_GETTY_UNIT,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        await asyncio.wait_for(proc.communicate(), timeout=5)
        return proc.returncode == 0
    except (asyncio.TimeoutError, FileNotFoundError, OSError):
        return False


async def _chvt(vt: str) -> bool:
    """Activate a virtual terminal.  Returns True on success."""
    try:
        proc = await asyncio.create_subprocess_exec(
            "/usr/bin/chvt", vt,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=10)
        if proc.returncode == 0:
            return True
        logger.warning("Failed to switch to VT %s: %s", vt, stderr.decode().strip())
        return False
    except asyncio.TimeoutError:
        logger.warning("Timeout switching to VT %s", vt)
        return False
    except FileNotFoundError:
        logger.warning("chvt not found; cannot switch VTs")
        return False
    except OSError as e:
        logger.warning("Failed to switch to VT %s: %s", vt, e)
        return False


def _write_console_message():
    """Print a short message on the console so the screen is not left blank."""
    try:
        with open(CONSOLE_DEVICE, "w") as console:
            console.write(CONSOLE_MESSAGE)
    except OSError as e:
        logger.warning("Could not write to %s: %s", CONSOLE_DEVICE, e)


async def _switch_to_console_vt() -> bool:
    """
    Leave the physical display showing a usable terminal.

    Best-effort: never fails the caller.  Returns True when the console was
    activated and repainted.
    """
    if not await _console_getty_active():
        # Without a getty there is no terminal to show, so leave the display alone.
        logger.warning("No getty on %s; leaving the display as-is", CONSOLE_GETTY_UNIT)
        return False

    if not await _chvt(CONSOLE_VT):
        return False

    # The console VT is usually already in the foreground, but stale after the
    # display manager released the display; bounce off another VT to repaint it.
    if await _chvt(REPAINT_VT) and not await _chvt(CONSOLE_VT):
        logger.error("Left the display on VT %s after failing to return to VT %s",
                     REPAINT_VT, CONSOLE_VT)
        return False

    _write_console_message()
    logger.info("Console VT %s activated and repainted", CONSOLE_VT)
    return True


async def start_desktop() -> dict:
    """
    Start the graphical desktop (display manager) until stopped or reboot.

    This deliberately does not `systemctl enable` the unit; whether the desktop
    starts on boot is left to the system's default target.

    Returns:
        dict with success status
    """
    return await _systemctl_desktop("start")


async def stop_desktop() -> dict:
    """
    Stop the graphical desktop (display manager) until started or reboot, then
    drop the physical display to a console terminal.

    Fail open: this deliberately does not `systemctl disable` the unit, so the
    desktop returns on the next boot.

    Returns:
        dict with success status and whether the display reached the console
    """
    result = await _systemctl_desktop("stop")
    if result.get("success"):
        result["switched_to_console"] = await _switch_to_console_vt()
    return result


# --- VNC ---

# Raspberry Pi OS ships wayvnc for Wayland sessions and RealVNC for X11; a Pi may
# have both installed.  wayvnc only works when its config exists, which is also
# the unit's own ConditionPathExists.
WAYVNC_UNIT = "wayvnc.service"
WAYVNC_CONFIG = Path("/etc/wayvnc/config")
REALVNC_UNIT = "vncserver-x11-serviced.service"
VNC_PORT = 5900


async def get_vnc_backends() -> tuple[list[dict], str | None]:
    """
    List the usable VNC backends in preference order, with their current state.

    Both stacks can be installed at once and *either* can be running — a user may
    have started RealVNC through raspi-config.  Reporting only the preferred
    backend would claim "VNC is off" while the other one is still serving the
    screen, so every installed backend is reported and `stop_vnc()` stops all of
    the running ones.

    Prefers wayvnc (Raspberry Pi OS uses Wayland) when choosing what to start.
    Deliberately does not copy `raspi-config`'s live-compositor probe
    (`pgrep labwc || pgrep wayfire`), which reports X11 — and would therefore pick
    RealVNC — whenever the desktop happens to be stopped.

    Never raises: a host without systemd simply has no VNC backend.

    Returns:
        (backends, reason) where each backend is {"unit", "name", "active"} and
        reason explains an empty list
    """
    unavailable = "VNC not supported or systemctl unavailable"
    try:
        properties = await _units_properties([WAYVNC_UNIT, REALVNC_UNIT], "LoadState", "ActiveState")
    except (asyncio.TimeoutError, OSError):
        return [], unavailable
    if properties is None:
        return [], unavailable

    wayvnc_properties, realvnc_properties = properties[0], properties[1]

    backends = []
    wayvnc_loaded = wayvnc_properties.get("LoadState") == "loaded"
    if wayvnc_loaded and WAYVNC_CONFIG.is_file():
        backends.append({
            "unit": WAYVNC_UNIT,
            "name": "wayvnc",
            "active": wayvnc_properties.get("ActiveState") == "active",
        })
    if realvnc_properties.get("LoadState") == "loaded":
        backends.append({
            "unit": REALVNC_UNIT,
            "name": "realvnc",
            "active": realvnc_properties.get("ActiveState") == "active",
        })

    if backends:
        return backends, None
    if wayvnc_loaded:
        # Installed but unconfigured; starting it would fail its own condition.
        return [], f"wayvnc is installed but not configured ({WAYVNC_CONFIG} is missing)"
    return [], "No VNC server installed"


async def get_vnc_status() -> VncStatus:
    """
    Get the VNC server status; on when *any* installed backend is running.

    Returns:
        VncStatus enum value
    """
    if is_docker_mode():
        return VncStatus.unknown

    backends, _ = await get_vnc_backends()
    if not backends:
        return VncStatus.unavailable

    return VncStatus.on if any(backend["active"] for backend in backends) else VncStatus.off


async def get_vnc_status_dict() -> dict:
    """
    Get VNC status as a dict for API responses.

    Returns dict matching VncStatusResponse schema:
        running: bool - Whether any VNC server is currently running
        available: bool - Whether a usable VNC server is installed
        desktop_running: bool - Whether the desktop VNC would serve is running
        can_start: bool - Whether VNC can be started right now
        reason: Optional[str] - Why VNC is unavailable or cannot be started
        backend: Optional[str] - The running backend, else the preferred one
        port: int - Port the VNC server listens on
    """
    if is_docker_mode():
        return {
            "running": False,
            "available": False,
            "desktop_running": False,
            "can_start": False,
            "reason": "Not available in Docker mode",
            "backend": None,
            "port": VNC_PORT,
        }

    backends, unavailable_reason = await get_vnc_backends()
    desktop_running = await get_desktop_status() == DesktopStatus.on

    active = [backend for backend in backends if backend["active"]]
    running = bool(active)
    available = bool(backends)
    # VNC serves the desktop session, so there is nothing to show without it.
    can_start = available and desktop_running
    # Name whichever backend is actually serving, so the UI cannot report wayvnc
    # while RealVNC is the one running.
    backend = (active or backends)[0]["name"] if backends else None

    reason = None
    if not available:
        reason = unavailable_reason or "No VNC server installed"
    elif not desktop_running:
        reason = "The Desktop must be running to use VNC"

    return {
        "running": running,
        "available": available,
        "desktop_running": desktop_running,
        "can_start": can_start,
        "reason": reason,
        "backend": backend,
        "port": VNC_PORT,
    }


async def start_vnc() -> dict:
    """
    Start the VNC server until stopped or reboot.

    Requires a running desktop: VNC serves that session, so starting it without
    one gives clients a blank screen.

    Fail open in the same sense as the desktop: the unit is never enabled, so VNC
    does not come back after a reboot.  Users who want it permanently on can
    enable it with raspi-config.

    Returns:
        dict with success status
    """
    if is_docker_mode():
        return {"success": False, "error": "Not available in Docker mode"}

    backends, reason = await get_vnc_backends()
    if not backends:
        return {"success": False, "error": reason or "No VNC server installed"}

    if await get_desktop_status() != DesktopStatus.on:
        return {
            "success": False,
            "error": "The Desktop must be running to start VNC",
            "precondition_failed": True,
        }

    return await _systemctl_unit("start", backends[0]["unit"], "VNC")


async def stop_vnc() -> dict:
    """
    Stop every running VNC server.

    All running backends are stopped, not just the preferred one: leaving the
    other stack serving while the Controller reports VNC as off would keep the
    screen remotely accessible.

    Always allowed, even with the desktop stopped, so a running VNC server can
    never be left on with no way to turn it off.

    Returns:
        dict with success status
    """
    if is_docker_mode():
        return {"success": False, "error": "Not available in Docker mode"}

    backends, reason = await get_vnc_backends()
    if not backends:
        return {"success": False, "error": reason or "No VNC server installed"}

    # Nothing running: still issue a stop so the call stays idempotent.
    units = [backend["unit"] for backend in backends if backend["active"]] or [backends[0]["unit"]]

    errors = []
    for unit in units:
        result = await _systemctl_unit("stop", unit, "VNC")
        if not result.get("success"):
            errors.append(f"{unit}: {result.get('error')}")

    if errors:
        return {"success": False, "error": "; ".join(errors)}
    return {"success": True}


GOVERNOR_MAP = {
    'ondemand': GovernorStatus.ondemand,
    'powersave': GovernorStatus.powersave,
}


def get_throttle_status() -> GovernorStatus:
    """
    Get CPU throttle (governor) status matching wrolpi/admin.py logic.

    Returns:
        GovernorStatus enum value
    """
    if is_docker_mode():
        return GovernorStatus.unknown

    governor_path = Path("/sys/devices/system/cpu/cpu0/cpufreq/scaling_governor")

    if not governor_path.exists():
        return GovernorStatus.unknown

    try:
        current = governor_path.read_text().strip()
        return GOVERNOR_MAP.get(current, GovernorStatus.unknown)
    except IOError:
        return GovernorStatus.unknown


def get_throttle_status_dict() -> dict:
    """
    Get throttle status as a dict for API responses.

    Returns dict matching ThrottleStatusResponse schema:
        enabled: bool - Whether CPU throttle (powersave) is enabled
        available: bool - Whether throttle control is available
        reason: Optional[str] - Reason if unavailable
        governor: Optional[str] - Current CPU governor
        available_governors: Optional[list[str]] - Available governors
    """
    if is_docker_mode():
        return {
            "enabled": False,
            "available": False,
            "reason": "Docker mode",
            "governor": None,
            "available_governors": None,
        }

    governor_path = Path("/sys/devices/system/cpu/cpu0/cpufreq/scaling_governor")
    available_path = Path("/sys/devices/system/cpu/cpu0/cpufreq/scaling_available_governors")

    if not governor_path.exists():
        return {
            "enabled": False,
            "available": False,
            "reason": "cpufreq not available",
            "governor": None,
            "available_governors": None,
        }

    try:
        current = governor_path.read_text().strip()
        available_governors = available_path.read_text().strip().split() if available_path.exists() else []

        return {
            "enabled": current == "powersave",
            "available": True,
            "reason": None,
            "governor": current,
            "available_governors": available_governors,
        }
    except IOError as e:
        return {
            "enabled": False,
            "available": False,
            "reason": str(e),
            "governor": None,
            "available_governors": None,
        }


def enable_throttle() -> dict:
    """
    Enable CPU throttle (set powersave governor).

    Returns:
        dict with success status
    """
    if is_docker_mode():
        return {"success": False, "error": "Not available in Docker mode"}

    logger.info("Enabling CPU throttle (powersave)")
    try:
        # Try cpufreq-set first
        result = subprocess.run(
            ["cpufreq-set", "-g", "powersave"],
            capture_output=True,
            text=True,
            timeout=5,
        )

        if result.returncode == 0:
            logger.info("CPU throttle enabled (powersave)")
            return {"success": True, "governor": "powersave"}
        else:
            logger.warning("Failed to enable CPU throttle: %s", result.stderr.strip())
            return {"success": False, "error": result.stderr.strip() or "Unknown error"}

    except FileNotFoundError:
        # Try direct write to sysfs
        try:
            for cpu_path in Path("/sys/devices/system/cpu/").glob("cpu[0-9]*/cpufreq/scaling_governor"):
                cpu_path.write_text("powersave")
            logger.info("CPU throttle enabled (powersave) via sysfs")
            return {"success": True, "governor": "powersave"}
        except IOError as e:
            logger.warning("Failed to enable CPU throttle via sysfs: %s", e)
            return {"success": False, "error": str(e)}
    except subprocess.TimeoutExpired:
        logger.warning("Timeout enabling CPU throttle")
        return {"success": False, "error": "Timeout enabling throttle"}
    except subprocess.SubprocessError as e:
        logger.warning("Failed to enable CPU throttle: %s", e)
        return {"success": False, "error": str(e)}


def disable_throttle() -> dict:
    """
    Disable CPU throttle (set ondemand or performance governor).

    Returns:
        dict with success status
    """
    if is_docker_mode():
        return {"success": False, "error": "Not available in Docker mode"}

    default_governor = DEFAULT_CPU_FREQUENCY

    logger.info("Disabling CPU throttle (setting governor to %s)", default_governor)
    try:
        result = subprocess.run(
            ["cpufreq-set", "-g", default_governor],
            capture_output=True,
            text=True,
            timeout=5,
        )

        if result.returncode == 0:
            logger.info("CPU throttle disabled (governor: %s)", default_governor)
            return {"success": True, "governor": default_governor}
        else:
            logger.warning("Failed to disable CPU throttle: %s", result.stderr.strip())
            return {"success": False, "error": result.stderr.strip() or "Unknown error"}

    except FileNotFoundError:
        # Try direct write to sysfs
        try:
            for cpu_path in Path("/sys/devices/system/cpu/").glob("cpu[0-9]*/cpufreq/scaling_governor"):
                cpu_path.write_text(default_governor)
            logger.info("CPU throttle disabled (governor: %s) via sysfs", default_governor)
            return {"success": True, "governor": default_governor}
        except IOError as e:
            logger.warning("Failed to disable CPU throttle via sysfs: %s", e)
            return {"success": False, "error": str(e)}
    except subprocess.TimeoutExpired:
        logger.warning("Timeout disabling CPU throttle")
        return {"success": False, "error": "Timeout disabling throttle"}
    except subprocess.SubprocessError as e:
        logger.warning("Failed to disable CPU throttle: %s", e)
        return {"success": False, "error": str(e)}


def shutdown_system(delay: int = 10) -> dict:
    """
    Shut down the system after a delay.

    Args:
        delay: Seconds to wait before shutdown (default 10). Allows API to return response.

    Returns:
        dict with success status
    """
    import shutil

    if is_docker_mode():
        return {"success": False, "error": "Cannot shutdown from Docker container"}

    # Check if shutdown command exists before starting thread
    if not shutil.which("shutdown"):
        return {"success": False, "error": "shutdown command not found"}

    def _delayed_shutdown():
        import time
        time.sleep(delay)
        subprocess.Popen(["shutdown", "-h", "now"])

    try:
        import threading
        thread = threading.Thread(target=_delayed_shutdown, daemon=True)
        thread.start()
        logger.info("System shutdown scheduled in %d seconds", delay)
        return {"success": True, "message": f"System shutting down in {delay} seconds"}
    except Exception as e:
        logger.warning("Failed to schedule system shutdown: %s", e)
        return {"success": False, "error": str(e)}


def reboot_system(delay: int = 10) -> dict:
    """
    Reboot the system after a delay.

    Args:
        delay: Seconds to wait before reboot (default 10). Allows API to return response.

    Returns:
        dict with success status
    """
    import shutil

    if is_docker_mode():
        return {"success": False, "error": "Cannot reboot from Docker container"}

    # Check if reboot command exists before starting thread
    if not shutil.which("reboot"):
        return {"success": False, "error": "reboot command not found"}

    def _delayed_reboot():
        import time
        time.sleep(delay)
        subprocess.Popen(["reboot"])

    try:
        import threading
        thread = threading.Thread(target=_delayed_reboot, daemon=True)
        thread.start()
        logger.info("System reboot scheduled in %d seconds", delay)
        return {"success": True, "message": f"System rebooting in {delay} seconds"}
    except Exception as e:
        logger.warning("Failed to schedule system reboot: %s", e)
        return {"success": False, "error": str(e)}


async def restart_all_services() -> dict:
    """
    Restart all WROLPi services including Controller itself.

    Controller restarts itself LAST so it can report success for other services.

    Returns:
        dict with results for each service
    """
    if is_docker_mode():
        # In Docker mode, we can't use systemctl
        return {
            "success": False,
            "error": "Service restart via systemctl not available in Docker mode",
        }

    services_to_restart = [
        "wrolpi-api",
        "wrolpi-app",
        "wrolpi-kiwix",
        "wrolpi-help",
        # Controller restarts itself last (scheduled after response)
    ]

    results = {}

    logger.info("Restarting all WROLPi services")
    for service in services_to_restart:
        logger.info("Restarting service %s", service)
        try:
            result = subprocess.run(
                ["systemctl", "restart", service],
                capture_output=True,
                text=True,
                timeout=30,
            )
            results[service] = {
                "success": result.returncode == 0,
                "error": result.stderr.strip() if result.returncode != 0 else None,
            }
            if result.returncode == 0:
                logger.info("Service %s restarted successfully", service)
            else:
                logger.warning("Failed to restart service %s: %s", service, result.stderr.strip())
        except subprocess.TimeoutExpired:
            logger.warning("Timeout restarting service %s", service)
            results[service] = {"success": False, "error": "Timeout"}
        except FileNotFoundError:
            logger.warning("systemctl not found while restarting %s", service)
            results[service] = {"success": False, "error": "systemctl not found"}
        except Exception as e:
            logger.warning("Failed to restart service %s: %s", service, e)
            results[service] = {"success": False, "error": str(e)}

    # Schedule Controller self-restart after response is sent
    loop = asyncio.get_event_loop()
    loop.call_later(1.0, _restart_self)

    results["wrolpi-controller"] = {"success": True, "pending": True}
    return {"success": True, "services": results}


def _restart_self():
    """Restart the Controller service. Called after response is sent."""
    try:
        subprocess.Popen(["systemctl", "restart", "wrolpi-controller"])
    except Exception:
        # If we can't restart ourselves, not much we can do
        pass


# --- Timezone ---


def get_timezone_status_dict() -> dict:
    """
    Get current system timezone status.

    Returns dict matching TimezoneStatusResponse schema:
        available: bool - Whether system timezone control is available
        timezone: Optional[str] - Current system timezone (IANA)
        reason: Optional[str] - Reason if unavailable
    """
    if is_docker_mode():
        return {
            "available": False,
            "timezone": None,
            "reason": "Docker mode",
        }

    try:
        result = subprocess.run(
            ["timedatectl", "show", "--property=Timezone", "--value"],
            capture_output=True,
            text=True,
            timeout=5,
        )
        if result.returncode == 0:
            return {
                "available": True,
                "timezone": result.stdout.strip(),
                "reason": None,
            }
        else:
            return {
                "available": False,
                "timezone": None,
                "reason": result.stderr.strip() or "timedatectl returned an error",
            }
    except FileNotFoundError:
        return {
            "available": False,
            "timezone": None,
            "reason": "timedatectl not found",
        }
    except subprocess.TimeoutExpired:
        return {
            "available": False,
            "timezone": None,
            "reason": "timedatectl timed out",
        }
    except subprocess.SubprocessError as e:
        return {
            "available": False,
            "timezone": None,
            "reason": str(e),
        }


def set_timezone(timezone: str) -> dict:
    """
    Set the system timezone using timedatectl.

    Args:
        timezone: IANA timezone string (e.g. "America/Denver")

    Returns:
        dict with success status
    """
    if is_docker_mode():
        return {"success": False, "timezone": None, "error": "Not available in Docker mode"}

    if not timezone or not timezone.strip():
        return {"success": False, "timezone": None, "error": "Timezone must not be empty"}

    logger.info("Setting system timezone to %s", timezone)
    try:
        result = subprocess.run(
            ["timedatectl", "set-timezone", timezone],
            capture_output=True,
            text=True,
            timeout=10,
        )

        if result.returncode == 0:
            logger.info("System timezone set to %s", timezone)
            return {"success": True, "timezone": timezone, "error": None}
        else:
            logger.warning("Failed to set timezone to %s: %s", timezone, result.stderr.strip())
            return {"success": False, "timezone": None, "error": result.stderr.strip() or "Unknown error"}

    except FileNotFoundError:
        logger.warning("timedatectl not found, cannot set timezone")
        return {"success": False, "timezone": None, "error": "timedatectl not found"}
    except subprocess.TimeoutExpired:
        logger.warning("Timeout setting timezone to %s", timezone)
        return {"success": False, "timezone": None, "error": "Timeout setting timezone"}
    except subprocess.SubprocessError as e:
        logger.warning("Failed to set timezone: %s", e)
        return {"success": False, "timezone": None, "error": str(e)}


def apply_timezone_from_config():
    """
    Read the timezone from wrolpi.yaml and apply it to the system.
    Called on Controller startup to ensure system timezone matches config.
    """
    if is_docker_mode():
        return

    config_path = get_media_directory() / 'config' / 'wrolpi.yaml'
    if not config_path.exists():
        return

    try:
        with open(config_path) as f:
            config = yaml.safe_load(f) or {}
    except (IOError, yaml.YAMLError) as e:
        logger.warning("Failed to read wrolpi.yaml for timezone: %s", e)
        return

    timezone = config.get('timezone')
    if not timezone:
        return

    result = set_timezone(timezone)
    if result.get('success'):
        logger.info("Applied system timezone from config: %s", timezone)
    else:
        logger.warning("Failed to apply system timezone '%s': %s", timezone, result.get('error'))