mirror of
https://github.com/lrnselfreliance/wrolpi.git
synced 2026-10-26 10:51:36 -04:00
3e559e1bf8
Add a Desktop toggle to the Controller (both UIs) that starts/stops display-manager.service. Fail open: the unit is never enabled/disabled, so the desktop always returns on reboot; users who want it disabled across reboots can set the default target themselves. Rename the hotspot and bluetooth actions to reflect their real behavior: - enable/disable_hotspot -> start/stop_hotspot (/api/hotspot/start|stop) - enable/disable_bluetooth -> unblock/block_bluetooth (rfkill semantics, /api/bluetooth/unblock|block) Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
1283 lines
44 KiB
Python
1283 lines
44 KiB
Python
"""
|
|
Admin operations for WROLPi Controller.
|
|
|
|
Migrated from wrolpi/admin.py - provides system control functions.
|
|
"""
|
|
|
|
import asyncio
|
|
import enum
|
|
import json
|
|
import logging
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
from controller.lib import config as config_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 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) -> dict:
|
|
"""
|
|
Update hotspot settings in controller.yaml. Only provided values are changed.
|
|
|
|
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 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)",
|
|
get_hotspot_device(), get_hotspot_ssid(), get_hotspot_protocol())
|
|
return {
|
|
"success": True,
|
|
"device": get_hotspot_device(),
|
|
"ssid": get_hotspot_ssid(),
|
|
"password": get_hotspot_password(),
|
|
"protocol": get_hotspot_protocol(),
|
|
}
|
|
|
|
|
|
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
|
|
|
|
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)
|
|
|
|
|
|
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
|
|
"""
|
|
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"
|
|
|
|
return {
|
|
"enabled": enabled,
|
|
"available": available,
|
|
"reason": reason,
|
|
"ssid": get_hotspot_ssid() if enabled else None,
|
|
"device": device,
|
|
}
|
|
|
|
|
|
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"}
|
|
|
|
# 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:
|
|
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)
|
|
|
|
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 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)")
|
|
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)}
|
|
|
|
|
|
# --- 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:
|
|
proc = await asyncio.create_subprocess_exec(
|
|
"systemctl", "show", DISPLAY_MANAGER_UNIT, "--property=LoadState,ActiveState",
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
)
|
|
stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=5)
|
|
|
|
if proc.returncode != 0:
|
|
return DesktopStatus.unknown
|
|
|
|
properties = dict(
|
|
line.split("=", 1) for line in stdout.decode().strip().splitlines() if "=" in line
|
|
)
|
|
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."""
|
|
if is_docker_mode():
|
|
return {"success": False, "error": "Not available in Docker mode"}
|
|
|
|
logger.info("Desktop %s requested (%s)", action, DISPLAY_MANAGER_UNIT)
|
|
try:
|
|
proc = await asyncio.create_subprocess_exec(
|
|
"systemctl", action, DISPLAY_MANAGER_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("Desktop %s succeeded", action)
|
|
return {"success": True}
|
|
else:
|
|
error_msg = stderr.decode().strip()
|
|
logger.warning("Failed to %s desktop: %s", action, error_msg)
|
|
return {"success": False, "error": error_msg or "Unknown error"}
|
|
|
|
except asyncio.TimeoutError:
|
|
logger.warning("Timeout during desktop %s", action)
|
|
return {"success": False, "error": f"Timeout during desktop {action}"}
|
|
except FileNotFoundError:
|
|
logger.warning("systemctl not found, cannot %s desktop", action)
|
|
return {"success": False, "error": "systemctl not found"}
|
|
except OSError as e:
|
|
logger.warning("Failed to %s desktop: %s", action, e)
|
|
return {"success": False, "error": str(e)}
|
|
|
|
|
|
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.
|
|
|
|
Fail open: this deliberately does not `systemctl disable` the unit, so the
|
|
desktop returns on the next boot.
|
|
|
|
Returns:
|
|
dict with success status
|
|
"""
|
|
return await _systemctl_desktop("stop")
|
|
|
|
|
|
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'))
|