need auth callback

pull/34814/head
deanlee 6 months ago committed by Cameron Clough
parent 9e8aa1d5e8
commit 29ab203a8b
  1. 11
      system/ui/lib/wifi_manager.py
  2. 17
      system/ui/network.py

@ -56,6 +56,7 @@ class WifiManager:
self.active_ap_path: str = '' self.active_ap_path: str = ''
self.scan_task: asyncio.Task | None = None self.scan_task: asyncio.Task | None = None
self.running: bool = True self.running: bool = True
self.need_auth_callback = None
async def connect(self): async def connect(self):
"""Connect to the DBus system bus.""" """Connect to the DBus system bus."""
@ -118,12 +119,8 @@ class WifiManager:
async def activate_connection(self, ssid: str) -> None: async def activate_connection(self, ssid: str) -> None:
connection_path = self.saved_connections.get(ssid) connection_path = self.saved_connections.get(ssid)
if connection_path: if connection_path:
print('activate connection:', connection_path) nm_iface = await self._get_interface(NM, NM_PATH, NM_IFACE)
introspection = await self.bus.introspect(NM, NM_PATH) await nm_iface.call_activate_connection(connection_path, self.device_path, '/')
proxy = self.bus.get_proxy_object(NM, NM_PATH, introspection)
interface = proxy.get_interface(NM_IFACE)
await interface.call_activate_connection(connection_path, self.device_path, '/')
async def connect_to_network(self, ssid: str, password: str = None, is_hidden: bool = False): async def connect_to_network(self, ssid: str, password: str = None, is_hidden: bool = False):
"""Connect to a selected WiFi network.""" """Connect to a selected WiFi network."""
@ -236,6 +233,8 @@ class WifiManager:
elif new_state in (NMDeviceState.DISCONNECTED, NMDeviceState.NEED_AUTH): elif new_state in (NMDeviceState.DISCONNECTED, NMDeviceState.NEED_AUTH):
for network in self.networks: for network in self.networks:
network.is_connected = False network.is_connected = False
if new_state == NMDeviceState.NEED_AUTH and self.need_auth_callback:
self.need_auth_callback()
def _on_new_connection(self, path: str) -> None: def _on_new_connection(self, path: str) -> None:
"""Callback for NewConnection signal.""" """Callback for NewConnection signal."""

@ -24,6 +24,7 @@ class ActionState(IntEnum):
class WifiManagerUI: class WifiManagerUI:
def __init__(self, wifi_manager): def __init__(self, wifi_manager):
self.wifi_manager = wifi_manager self.wifi_manager = wifi_manager
self.wifi_manager.need_auth_callback = self._need_auth
self._selected_network = None self._selected_network = None
self.item_height = 160 self.item_height = 160
self.btn_width = 200 self.btn_width = 200
@ -36,7 +37,6 @@ class WifiManagerUI:
async def _initialize(self) -> None: async def _initialize(self) -> None:
try: try:
await self.wifi_manager.connect() await self.wifi_manager.connect()
self.wifi_manager.bus.add_message_handler(self._handle_dbus_signal)
except Exception as e: except Exception as e:
print(f"Initialization error: {e}") print(f"Initialization error: {e}")
@ -51,9 +51,11 @@ class WifiManagerUI:
result = self.keyboard.render(rect, 'Enter password', f'for {self._selected_network.ssid}') result = self.keyboard.render(rect, 'Enter password', f'for {self._selected_network.ssid}')
if result == 0: if result == 0:
return return
else: elif result == 1:
self.current_action = ActionState.NONE self.current_action = ActionState.NONE
asyncio.create_task(self.connect_to_network(self.keyboard.text)) asyncio.create_task(self.connect_to_network(self.keyboard.text))
else:
self.current_action = ActionState.NONE
content_rect = rl.Rectangle( content_rect = rl.Rectangle(
rect.x, rect.y, rect.width, len(self.wifi_manager.networks) * self.item_height rect.x, rect.y, rect.width, len(self.wifi_manager.networks) * self.item_height
@ -128,15 +130,8 @@ class WifiManagerUI:
await self.wifi_manager.connect_to_network(self._selected_network.ssid, password) await self.wifi_manager.connect_to_network(self._selected_network.ssid, password)
self.current_action = ActionState.NONE self.current_action = ActionState.NONE
def _handle_dbus_signal(self, message): def _need_auth(self):
if message.message_type != MessageType.SIGNAL: self.current_action = ActionState.NEED_AUTH
return
if message.member == 'StateChanged':
if len(message.body) >= 2:
_, new_state = message.body[0], message.body[1]
if new_state == NM_DEVICE_STATE_NEED_AUTH:
self.current_action = ActionState.NEED_AUTH
async def main(): async def main():

Loading…
Cancel
Save