|
|
@ -57,9 +57,6 @@ class WifiManager: |
|
|
|
self.scan_task: asyncio.Task | None = None |
|
|
|
self.scan_task: asyncio.Task | None = None |
|
|
|
self.running: bool = True |
|
|
|
self.running: bool = True |
|
|
|
|
|
|
|
|
|
|
|
def is_saved(self, ssid: str) -> bool: |
|
|
|
|
|
|
|
return ssid in self.saved_connections |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
async def connect(self): |
|
|
|
async def connect(self): |
|
|
|
"""Connect to the DBus system bus.""" |
|
|
|
"""Connect to the DBus system bus.""" |
|
|
|
try: |
|
|
|
try: |
|
|
@ -110,7 +107,6 @@ class WifiManager: |
|
|
|
try: |
|
|
|
try: |
|
|
|
nm_iface = await self._get_interface(NM, path, NM_CONNECTION_IFACE) |
|
|
|
nm_iface = await self._get_interface(NM, path, NM_CONNECTION_IFACE) |
|
|
|
await nm_iface.call_delete() |
|
|
|
await nm_iface.call_delete() |
|
|
|
self.saved_connections.pop(ssid) |
|
|
|
|
|
|
|
return True |
|
|
|
return True |
|
|
|
except DBusError as e: |
|
|
|
except DBusError as e: |
|
|
|
cloudlog.error(f"Failed to delete connection for SSID: {ssid}. Error: {e}") |
|
|
|
cloudlog.error(f"Failed to delete connection for SSID: {ssid}. Error: {e}") |
|
|
@ -122,7 +118,7 @@ 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: |
|
|
|
cloudlog.info('activate connection:', connection_path) |
|
|
|
print('activate connection:', connection_path) |
|
|
|
introspection = await self.bus.introspect(NM, NM_PATH) |
|
|
|
introspection = await self.bus.introspect(NM, NM_PATH) |
|
|
|
proxy = self.bus.get_proxy_object(NM, NM_PATH, introspection) |
|
|
|
proxy = self.bus.get_proxy_object(NM, NM_PATH, introspection) |
|
|
|
interface = proxy.get_interface(NM_IFACE) |
|
|
|
interface = proxy.get_interface(NM_IFACE) |
|
|
@ -132,7 +128,7 @@ class WifiManager: |
|
|
|
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.""" |
|
|
|
try: |
|
|
|
try: |
|
|
|
settings_iface = await self._get_interface(NM, NM_SETTINGS_PATH, NM_SETTINGS_IFACE) |
|
|
|
# settings_iface = await self._get_interface(NM, NM_SETTINGS_PATH, NM_SETTINGS_IFACE) |
|
|
|
connection = { |
|
|
|
connection = { |
|
|
|
'connection': { |
|
|
|
'connection': { |
|
|
|
'type': Variant('s', '802-11-wireless'), |
|
|
|
'type': Variant('s', '802-11-wireless'), |
|
|
@ -159,7 +155,14 @@ class WifiManager: |
|
|
|
'psk': Variant('s', password), |
|
|
|
'psk': Variant('s', password), |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
await settings_iface.call_add_connection(connection) |
|
|
|
# nm_iface = self._get_interface(NM, NM_PATH, NM_IFACE) |
|
|
|
|
|
|
|
# await nm_iface.call_add_and_activate_connection(connection, self.device_path, '/') |
|
|
|
|
|
|
|
# await settings_iface.call_add_connection(connection) |
|
|
|
|
|
|
|
# introspection = await self.bus.introspect(NM, NM_PATH) |
|
|
|
|
|
|
|
# nm_proxy = self.bus.get_proxy_object(NM, NM_PATH, introspection) |
|
|
|
|
|
|
|
nm_iface = await self._get_interface(NM, NM_PATH, NM_IFACE) |
|
|
|
|
|
|
|
result = await nm_iface.call_add_and_activate_connection(connection, self.device_path, "/") |
|
|
|
|
|
|
|
print(result) |
|
|
|
|
|
|
|
|
|
|
|
for network in self.networks: |
|
|
|
for network in self.networks: |
|
|
|
network.is_connected = True if network.ssid == ssid else False |
|
|
|
network.is_connected = True if network.ssid == ssid else False |
|
|
@ -167,6 +170,9 @@ class WifiManager: |
|
|
|
except DBusError as e: |
|
|
|
except DBusError as e: |
|
|
|
cloudlog.error(f"Error connecting to network: {e}") |
|
|
|
cloudlog.error(f"Error connecting to network: {e}") |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def is_saved(self, ssid: str) -> bool: |
|
|
|
|
|
|
|
return ssid in self.saved_connections |
|
|
|
|
|
|
|
|
|
|
|
async def _find_wifi_device(self) -> bool: |
|
|
|
async def _find_wifi_device(self) -> bool: |
|
|
|
nm_iface = await self._get_interface(NM, NM_PATH, NM_IFACE) |
|
|
|
nm_iface = await self._get_interface(NM, NM_PATH, NM_IFACE) |
|
|
|
devices = await nm_iface.get_devices() |
|
|
|
devices = await nm_iface.get_devices() |
|
|
@ -295,7 +301,6 @@ class WifiManager: |
|
|
|
continue |
|
|
|
continue |
|
|
|
|
|
|
|
|
|
|
|
bssid = properties.get('HwAddress', Variant('s', '')).value |
|
|
|
bssid = properties.get('HwAddress', Variant('s', '')).value |
|
|
|
print(bssid) |
|
|
|
|
|
|
|
flags = properties['Flags'].value |
|
|
|
flags = properties['Flags'].value |
|
|
|
wpa_flags = properties['WpaFlags'].value |
|
|
|
wpa_flags = properties['WpaFlags'].value |
|
|
|
rsn_flags = properties['RsnFlags'].value |
|
|
|
rsn_flags = properties['RsnFlags'].value |
|
|
|