mawei/StarPilot
1
0
Fork 0
mirror of https://github.com/firestar5683/StarPilot.git synced 2026-06-30 02:52:04 +08:00
Code Issues Packages Projects Releases Wiki Activity

system/ui: fix remaining issues in WiFi Manager (#35301)

Browse Source 
* WIP

* fix callback

* fix connecting network displayed as Connected

* thread safe states

* fix state sync issues

* fix callback
This commit is contained in:
Dean Lee
2025-05-28 06:01:53 +08:00
committed by GitHub
parent 28da563386
commit 3a7f0b66aa
2 changed files with 93 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files
system/ui/lib/wifi_manager.py
+37 -32
View File
@@ -69,8 +69,9 @@ class NetworkInfo:
class WifiManagerCallbacks:
need_auth: Callable[[str], None] | None = None
activated: Callable[[], None] | None = None
forgotten: Callable[[], None] | None = None
forgotten: Callable[[str], None] | None = None
networks_updated: Callable[[list[NetworkInfo]], None] | None = None
connection_failed: Callable[[str, str], None] | None = None # Added for error feedback
class WifiManager:
@@ -98,8 +99,8 @@ class WifiManager:
self.bus = await MessageBus(bus_type=BusType.SYSTEM).connect()
if not await self._find_wifi_device():
raise ValueError("No Wi-Fi device found")
await self._setup_signals(self.device_path)
await self._setup_signals(self.device_path)
self.active_ap_path = await self.get_active_access_point()
await self.add_tethering_connection(self._tethering_ssid, DEFAULT_TETHERING_PASSWORD)
self.saved_connections = await self._get_saved_connections()
@@ -122,7 +123,7 @@ class WifiManager:
if self.bus:
self.bus.disconnect()
async def request_scan(self) -> None:
async def _request_scan(self) -> None:
try:
interface = self.device_proxy.get_interface(NM_WIRELESS_IFACE)
await interface.call_request_scan({})
@@ -146,12 +147,23 @@ class WifiManager:
try:
nm_iface = await self._get_interface(NM, path, NM_CONNECTION_IFACE)
await nm_iface.call_delete()
if self._current_connection_ssid == ssid:
self._current_connection_ssid = None
if ssid in self.saved_connections:
del self.saved_connections[ssid]
for network in self.networks:
if network.ssid == ssid:
network.is_saved = False
network.is_connected = False
break
# Notify UI of forgotten connection
if self.callbacks.networks_updated:
self.callbacks.networks_updated(copy.deepcopy(self.networks))
return True
except DBusError as e:
cloudlog.error(f"Failed to delete connection for SSID: {ssid}. Error: {e}")
@@ -212,10 +224,12 @@ class WifiManager:
nm_iface = await self._get_interface(NM, NM_PATH, NM_IFACE)
await nm_iface.call_add_and_activate_connection(connection, self.device_path, "/")
await self._update_connection_status()
except DBusError as e:
except Exception as e:
self._current_connection_ssid = None
cloudlog.error(f"Error connecting to network: {e}")
# Notify UI of failure
if self.callbacks.connection_failed:
self.callbacks.connection_failed(ssid, str(e))
def is_saved(self, ssid: str) -> bool:
return ssid in self.saved_connections
@@ -393,8 +407,7 @@ class WifiManager:
async def _periodic_scan(self):
while self.running:
try:
await self.request_scan()
await self._get_available_networks()
await self._request_scan()
await asyncio.sleep(30)
except asyncio.CancelledError:
break
@@ -423,21 +436,24 @@ class WifiManager:
def _on_properties_changed(self, interface: str, changed: dict, invalidated: list):
# print("property changed", interface, changed, invalidated)
if 'LastScan' in changed:
asyncio.create_task(self._get_available_networks())
asyncio.create_task(self._refresh_networks())
elif interface == NM_WIRELESS_IFACE and "ActiveAccessPoint" in changed:
self.active_ap_path = changed["ActiveAccessPoint"].value
asyncio.create_task(self._get_available_networks())
new_ap_path = changed["ActiveAccessPoint"].value
if self.active_ap_path != new_ap_path:
self.active_ap_path = new_ap_path
asyncio.create_task(self._refresh_networks())
def _on_state_changed(self, new_state: int, old_state: int, reason: int):
print(f"State changed: {old_state} -> {new_state}, reason: {reason}")
print("State changed", new_state, old_state, reason)
if new_state == NMDeviceState.ACTIVATED:
if self.callbacks.activated:
self.callbacks.activated()
asyncio.create_task(self._update_connection_status())
asyncio.create_task(self._refresh_networks())
self._current_connection_ssid = None
elif new_state in (NMDeviceState.DISCONNECTED, NMDeviceState.NEED_AUTH):
for network in self.networks:
network.is_connected = False
if new_state == NMDeviceState.NEED_AUTH and reason == NM_DEVICE_STATE_REASON_SUPPLICANT_DISCONNECT and self.callbacks.need_auth:
if self._current_connection_ssid:
self.callbacks.need_auth(self._current_connection_ssid)
@@ -453,19 +469,19 @@ class WifiManager:
def _on_new_connection(self, path: str) -> None:
"""Callback for NewConnection signal."""
print(f"New connection added: {path}")
asyncio.create_task(self._add_saved_connection(path))
def _on_connection_removed(self, path: str) -> None:
"""Callback for ConnectionRemoved signal."""
print(f"Connection removed: {path}")
for ssid, p in list(self.saved_connections.items()):
if path == p:
del self.saved_connections[ssid]
if self.callbacks.forgotten:
self.callbacks.forgotten()
self.callbacks.forgotten(ssid)
# Update network list to reflect the removed saved connection
asyncio.create_task(self._update_connection_status())
asyncio.create_task(self._refresh_networks())
break
async def _add_saved_connection(self, path: str) -> None:
@@ -474,7 +490,7 @@ class WifiManager:
settings = await self._get_connection_settings(path)
if ssid := self._extract_ssid(settings):
self.saved_connections[ssid] = path
await self._update_connection_status()
await self._refresh_networks()
except DBusError as e:
cloudlog.error(f"Failed to add connection {path}: {e}")
@@ -483,10 +499,6 @@ class WifiManager:
ssid_variant = settings.get('802-11-wireless', {}).get('ssid', Variant('ay', b'')).value
return ''.join(chr(b) for b in ssid_variant) if ssid_variant else None
async def _update_connection_status(self):
self.active_ap_path = await self.get_active_access_point()
await self._get_available_networks()
async def _add_match_rule(self, rule):
"""Add a match rule on the bus."""
reply = await self.bus.call(
@@ -504,10 +516,11 @@ class WifiManager:
assert reply.message_type == MessageType.METHOD_RETURN
return reply
async def _get_available_networks(self):
async def _refresh_networks(self):
"""Get a list of available networks via NetworkManager."""
wifi_iface = self.device_proxy.get_interface(NM_WIRELESS_IFACE)
access_points = await wifi_iface.get_access_points()
self.active_ap_path = await self.get_active_access_point()
network_dict = {}
for ap_path in access_points:
try:
@@ -531,7 +544,7 @@ class WifiManager:
security_type=self._get_security_type(flags, wpa_flags, rsn_flags),
path=ap_path,
bssid=bssid,
is_connected=self.active_ap_path == ap_path,
is_connected=self.active_ap_path == ap_path and self._current_connection_ssid != ssid,
is_saved=ssid in self.saved_connections
)
@@ -555,9 +568,7 @@ class WifiManager:
async def _get_connection_settings(self, path):
"""Fetch connection settings for a specific connection path."""
try:
connection_proxy = await self.bus.introspect(NM, path)
connection = self.bus.get_proxy_object(NM, path, connection_proxy)
settings = connection.get_interface(NM_CONNECTION_IFACE)
settings = await self._get_interface(NM, path, NM_CONNECTION_IFACE)
return await settings.call_get_settings()
except DBusError as e:
cloudlog.error(f"Failed to get settings for {path}: {str(e)}")
@@ -660,12 +671,6 @@ class WifiManagerWrapper:
return
self._run_coroutine(self._manager.connect())
def request_scan(self):
"""Request a scan for Wi-Fi networks."""
if not self._manager:
return
self._run_coroutine(self._manager.request_scan())
def forget_connection(self, ssid: str):
"""Forget a saved Wi-Fi connection."""
if not self._manager:
system/ui/widgets/network.py
+56 -39
View File
@@ -1,4 +1,5 @@
from dataclasses import dataclass
from threading import Lock
from typing import Literal
import pyray as rl
@@ -53,48 +54,59 @@ UIState = StateIdle | StateConnecting | StateNeedsAuth | StateShowForgetConfirm
class WifiManagerUI:
def __init__(self, wifi_manager: WifiManagerWrapper):
self.state: UIState = StateIdle()
self.btn_width = 200
self.btn_width: int = 200
self.scroll_panel = GuiScrollPanel()
self.keyboard = Keyboard(max_text_size=MAX_PASSWORD_LENGTH, min_text_size=MIN_PASSWORD_LENGTH, show_password_toggle=True)
self._networks: list[NetworkInfo] = []
self._lock = Lock()
self.wifi_manager = wifi_manager
self.wifi_manager.set_callbacks(WifiManagerCallbacks(self._on_need_auth, self._on_activated, self._on_forgotten, self._on_network_updated))
self.wifi_manager.set_callbacks(
WifiManagerCallbacks(
need_auth = self._on_need_auth,
activated = self._on_activated,
forgotten = self._on_forgotten,
networks_updated = self._on_network_updated,
connection_failed = self._on_connection_failed
)
)
self.wifi_manager.start()
self.wifi_manager.connect()
def render(self, rect: rl.Rectangle):
if not self._networks:
gui_label(rect, "Scanning Wi-Fi networks...", 72, alignment=rl.GuiTextAlignment.TEXT_ALIGN_CENTER)
return
with self._lock:
if not self._networks:
gui_label(rect, "Scanning Wi-Fi networks...", 72, alignment=rl.GuiTextAlignment.TEXT_ALIGN_CENTER)
return
match self.state:
case StateNeedsAuth(network):
result = self.keyboard.render("Enter password", f"for {network.ssid}")
if result == 1:
password = self.keyboard.text
self.keyboard.clear()
match self.state:
case StateNeedsAuth(network):
result = self.keyboard.render("Enter password", f"for {network.ssid}")
if result == 1:
password = self.keyboard.text
self.keyboard.clear()
if len(password) >= MIN_PASSWORD_LENGTH:
self.connect_to_network(network, password)
elif result == 0:
self.state = StateIdle()
if len(password) >= MIN_PASSWORD_LENGTH:
self.connect_to_network(network, password)
elif result == 0:
self.state = StateIdle()
case StateShowForgetConfirm(network):
result = confirm_dialog(f'Forget Wi-Fi Network "{network.ssid}"?', "Forget")
if result == 1:
self.forget_network(network)
elif result == 0:
self.state = StateIdle()
case StateShowForgetConfirm(network):
result = confirm_dialog(f'Forget Wi-Fi Network "{network.ssid}"?', "Forget")
if result == 1:
self.forget_network(network)
elif result == 0:
self.state = StateIdle()
case _:
self._draw_network_list(rect)
case _:
self._draw_network_list(rect)
@property
def require_full_screen(self) -> bool:
"""Check if the WiFi UI requires exclusive full-screen rendering."""
return isinstance(self.state, (StateNeedsAuth, StateShowForgetConfirm))
with self._lock:
return isinstance(self.state, (StateNeedsAuth, StateShowForgetConfirm))
def _draw_network_list(self, rect: rl.Rectangle):
content_rect = rl.Rectangle(rect.x, rect.y, rect.width, len(self._networks) * ITEM_HEIGHT)
@@ -190,25 +202,30 @@ class WifiManagerUI:
self.wifi_manager.forget_connection(network.ssid)
def _on_network_updated(self, networks: list[NetworkInfo]):
self._networks = networks
with self._lock:
self._networks = networks
def _on_need_auth(self, ssid):
match self.state:
case StateConnecting(ssid):
self.state = StateNeedsAuth(ssid)
case _:
# Find network by SSID
network = next((n for n in self.wifi_manager.networks if n.ssid == ssid), None)
if network:
self.state = StateNeedsAuth(network)
with self._lock:
network = next((n for n in self._networks if n.ssid == ssid), None)
if network:
self.state = StateNeedsAuth(network)
def _on_activated(self):
if isinstance(self.state, StateConnecting):
self.state = StateIdle()
with self._lock:
if isinstance(self.state, StateConnecting):
self.state = StateIdle()
def _on_forgotten(self, ssid):
with self._lock:
if isinstance(self.state, StateForgetting):
self.state = StateIdle()
def _on_connection_failed(self, ssid: str, error: str):
with self._lock:
if isinstance(self.state, StateConnecting):
self.state = StateIdle()
def _on_forgotten(self):
if isinstance(self.state, StateForgetting):
self.state = StateIdle()
def main():