|
|
@ -59,8 +59,8 @@ class WifiManagerCallbacks: |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class WifiManager: |
|
|
|
class WifiManager: |
|
|
|
def __init__(self, callbacks: WifiManagerCallbacks): |
|
|
|
def __init__(self, callbacks): |
|
|
|
self.callbacks = callbacks |
|
|
|
self.callbacks: WifiManagerCallbacks = callbacks |
|
|
|
self.networks: list[NetworkInfo] = [] |
|
|
|
self.networks: list[NetworkInfo] = [] |
|
|
|
self.bus: MessageBus = None |
|
|
|
self.bus: MessageBus = None |
|
|
|
self.device_path: str = "" |
|
|
|
self.device_path: str = "" |
|
|
@ -394,24 +394,18 @@ class WifiManagerWrapper: |
|
|
|
self._manager: WifiManager | None = None |
|
|
|
self._manager: WifiManager | None = None |
|
|
|
self._callbacks: WifiManagerCallbacks = WifiManagerCallbacks() |
|
|
|
self._callbacks: WifiManagerCallbacks = WifiManagerCallbacks() |
|
|
|
|
|
|
|
|
|
|
|
self._loop = None |
|
|
|
|
|
|
|
self._running = False |
|
|
|
|
|
|
|
self._lock = threading.RLock() |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
self._thread = threading.Thread(target=self._run, daemon=True) |
|
|
|
self._thread = threading.Thread(target=self._run, daemon=True) |
|
|
|
self._thread.start() |
|
|
|
self._loop: asyncio.EventLoop | None = None |
|
|
|
|
|
|
|
self._running = False |
|
|
|
while self._thread is not None and not self._running: |
|
|
|
|
|
|
|
time.sleep(0.1) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@property |
|
|
|
def set_callbacks(self, callbacks: WifiManagerCallbacks): |
|
|
|
def callbacks(self) -> WifiManagerCallbacks: |
|
|
|
self._callbacks = callbacks |
|
|
|
return self._callbacks |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@callbacks.setter |
|
|
|
def start(self) -> None: |
|
|
|
def callbacks(self, callbacks: WifiManagerCallbacks): |
|
|
|
if not self._running: |
|
|
|
with self._lock: |
|
|
|
self._thread.start() |
|
|
|
self._callbacks = callbacks |
|
|
|
while self._thread is not None and not self._running: |
|
|
|
|
|
|
|
time.sleep(0.1) |
|
|
|
|
|
|
|
|
|
|
|
def _run(self): |
|
|
|
def _run(self): |
|
|
|
self._loop = asyncio.new_event_loop() |
|
|
|
self._loop = asyncio.new_event_loop() |
|
|
@ -428,9 +422,10 @@ class WifiManagerWrapper: |
|
|
|
self._loop.stop() |
|
|
|
self._loop.stop() |
|
|
|
self._running = False |
|
|
|
self._running = False |
|
|
|
|
|
|
|
|
|
|
|
def shutdown(self): |
|
|
|
def shutdown(self) -> None: |
|
|
|
if self._running: |
|
|
|
if self._running: |
|
|
|
self._run_coroutine(self._manager.shutdown()) |
|
|
|
if self._manager is not None: |
|
|
|
|
|
|
|
self._run_coroutine(self._manager.shutdown()) |
|
|
|
if self._loop and self._loop.is_running(): |
|
|
|
if self._loop and self._loop.is_running(): |
|
|
|
self._loop.call_soon_threadsafe(self._loop.stop) |
|
|
|
self._loop.call_soon_threadsafe(self._loop.stop) |
|
|
|
if self._thread and self._thread.is_alive(): |
|
|
|
if self._thread and self._thread.is_alive(): |
|
|
@ -439,33 +434,41 @@ class WifiManagerWrapper: |
|
|
|
|
|
|
|
|
|
|
|
@property |
|
|
|
@property |
|
|
|
def networks(self) -> list[NetworkInfo]: |
|
|
|
def networks(self) -> list[NetworkInfo]: |
|
|
|
"""Get the current list of networks (thread-safe).""" |
|
|
|
"""Get the current list of networks.""" |
|
|
|
with self._lock: |
|
|
|
return self._manager.networks if self._manager else [] |
|
|
|
return self._manager.networks if self._manager else [] |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def is_saved(self, ssid: str) -> bool: |
|
|
|
def is_saved(self, ssid: str) -> bool: |
|
|
|
"""Check if a network is saved (thread-safe).""" |
|
|
|
"""Check if a network is saved.""" |
|
|
|
with self._lock: |
|
|
|
return self._manager.is_saved(ssid) if self._manager else False |
|
|
|
return self._manager.is_saved(ssid) if self._manager else False |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def connect(self): |
|
|
|
def connect(self): |
|
|
|
"""Connect to DBus and start Wi-Fi scanning.""" |
|
|
|
"""Connect to DBus and start Wi-Fi scanning.""" |
|
|
|
|
|
|
|
if not self._manager: |
|
|
|
|
|
|
|
return |
|
|
|
self._run_coroutine(self._manager.connect()) |
|
|
|
self._run_coroutine(self._manager.connect()) |
|
|
|
|
|
|
|
|
|
|
|
def request_scan(self): |
|
|
|
def request_scan(self): |
|
|
|
"""Request a scan for Wi-Fi networks.""" |
|
|
|
"""Request a scan for Wi-Fi networks.""" |
|
|
|
|
|
|
|
if not self._manager: |
|
|
|
|
|
|
|
return |
|
|
|
self._run_coroutine(self._manager.request_scan()) |
|
|
|
self._run_coroutine(self._manager.request_scan()) |
|
|
|
|
|
|
|
|
|
|
|
def forget_connection(self, ssid: str): |
|
|
|
def forget_connection(self, ssid: str): |
|
|
|
"""Forget a saved Wi-Fi connection.""" |
|
|
|
"""Forget a saved Wi-Fi connection.""" |
|
|
|
|
|
|
|
if not self._manager: |
|
|
|
|
|
|
|
return |
|
|
|
self._run_coroutine(self._manager.forget_connection(ssid)) |
|
|
|
self._run_coroutine(self._manager.forget_connection(ssid)) |
|
|
|
|
|
|
|
|
|
|
|
def activate_connection(self, ssid: str): |
|
|
|
def activate_connection(self, ssid: str): |
|
|
|
"""Activate an existing Wi-Fi connection.""" |
|
|
|
"""Activate an existing Wi-Fi connection.""" |
|
|
|
|
|
|
|
if not self._manager: |
|
|
|
|
|
|
|
return |
|
|
|
self._run_coroutine(self._manager.activate_connection(ssid)) |
|
|
|
self._run_coroutine(self._manager.activate_connection(ssid)) |
|
|
|
|
|
|
|
|
|
|
|
def connect_to_network(self, ssid: str, password: str = None, bssid: str = None, is_hidden: bool = False): |
|
|
|
def connect_to_network(self, ssid: str, password: str = None, bssid: str = None, is_hidden: bool = False): |
|
|
|
"""Connect to a Wi-Fi network.""" |
|
|
|
"""Connect to a Wi-Fi network.""" |
|
|
|
|
|
|
|
if not self._manager: |
|
|
|
|
|
|
|
return |
|
|
|
self._run_coroutine(self._manager.connect_to_network(ssid, password, bssid, is_hidden)) |
|
|
|
self._run_coroutine(self._manager.connect_to_network(ssid, password, bssid, is_hidden)) |
|
|
|
|
|
|
|
|
|
|
|
def _run_coroutine(self, coro): |
|
|
|
def _run_coroutine(self, coro): |
|
|
|