network: new NetworkManager backend using sdbus (#1269)

* network: new NetworkManager backend using sdbus

drop support for old methods

* rescan changes

* add wifi on-off
This commit is contained in:
Alfredo Monclus 2024-05-24 19:13:35 -03:00 committed by GitHub
parent 524aa0e7dc
commit a80b881015
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
9 changed files with 457 additions and 2199 deletions

View File

@ -5,35 +5,10 @@
The network panel requires network-manager to function, (if you are using a fork this may not be the case)
Check if network-manager is installed:
```bash
dpkg -s network-manager
```
if the response is the following:
```sh
dpkg-query: the package 'network-manager' is not installed
```
if the response is the following:
```sh
Package: network-manager
Status: install ok installed
```
this line may appear in KlipperScreen.log:
!!! abstract "Log"
```sh
[wifi_nm.py:rescan()] [...] NetworkManager.wifi.scan request failed: not authorized
```
if version of KlipperScreen installed was previous than v0.3.9, then re-run the installer and reboot
??? Alternative workaround for network-manager
??? "Alternative workaround for network-manager not having permissions"
in order to fix this polkit needs to be configured or disabled:

View File

@ -1,27 +0,0 @@
## wpa_supplicant
The user is not allowed to control the interface
* Edit `/etc/wpa_supplicant/wpa_supplicant.conf` and add this line if it's not there:
```
ctrl_interface=DIR=/var/run/wpa_supplicant GROUP=netdev
```
* Run `cat /etc/group | grep netdev`
If your username is not listed under that line, you need to add it with the following command:
```sh
usermod -a -G netdev pi
```
(if your username is not 'pi' change 'pi' to your username)
Then reboot the machine:
```sh
systemctl reboot
```
!!! tip
It's possible to just restart KlipperScreen and networking

File diff suppressed because it is too large Load Diff

View File

@ -168,6 +168,15 @@ class ScreenPanel:
if size < unit:
return f"{(1024 * size / unit):.1f} {suffix}"
@staticmethod
def format_speed(bitrate):
bitrate = float(bitrate)
suffixes = ["Kbits/s", "Mbits/s", "Gbits/s", "Tbits/s", "Pbits/s", "Ebits/s", "Zbits/s", "Ybits/s"]
for i, suffix in enumerate(suffixes, start=1):
unit = 1000 ** i
if bitrate < unit:
return f"{(1000 * bitrate / unit):.0f} {suffix}"
@staticmethod
def prettify(name: str):
name = name.replace("_", " ")

250
ks_includes/sdbus_nm.py Normal file
View File

@ -0,0 +1,250 @@
# This is the backend of the UI panel that communicates to sdbus-networkmanager
# TODO device selection/swtichability
# Alfredo Monclus (alfrix) 2024
import logging
from sdbus_block.networkmanager import (
NetworkManager,
NetworkDeviceGeneric,
NetworkDeviceWireless,
NetworkConnectionSettings,
NetworkManagerSettings,
AccessPoint,
NetworkManagerConnectionProperties,
IPv4Config,
ActiveConnection,
enums,
)
from sdbus import sd_bus_open_system, set_default_bus
from gi.repository import GLib
from uuid import uuid4
NM_802_11_AP_SEC_NONE = 0
NM_802_11_AP_SEC_PAIR_WEP40 = 1
NM_802_11_AP_SEC_PAIR_WEP104 = 2
NM_802_11_AP_SEC_PAIR_TKIP = 4
NM_802_11_AP_SEC_PAIR_CCMP = 8
NM_802_11_AP_SEC_GROUP_WEP40 = 16
NM_802_11_AP_SEC_GROUP_WEP104 = 32
NM_802_11_AP_SEC_GROUP_TKIP = 64
NM_802_11_AP_SEC_GROUP_CCMP = 128
NM_802_11_AP_SEC_KEY_MGMT_PSK = 256
NM_802_11_AP_SEC_KEY_MGMT_802_1X = 512
def get_encryption(flags):
encryption = ""
if (flags & NM_802_11_AP_SEC_PAIR_WEP40 or
flags & NM_802_11_AP_SEC_PAIR_WEP104 or
flags & NM_802_11_AP_SEC_GROUP_WEP40 or
flags & NM_802_11_AP_SEC_GROUP_WEP104):
encryption += "WEP "
if (flags & NM_802_11_AP_SEC_PAIR_TKIP or
flags & NM_802_11_AP_SEC_GROUP_TKIP):
encryption += "TKIP "
if (flags & NM_802_11_AP_SEC_PAIR_CCMP or
flags & NM_802_11_AP_SEC_GROUP_CCMP):
encryption += "AES "
if flags & NM_802_11_AP_SEC_KEY_MGMT_PSK:
encryption += "WPA-PSK "
if flags & NM_802_11_AP_SEC_KEY_MGMT_802_1X:
encryption += "802.1x "
return encryption
def WifiChannels(freq: str):
if freq == '2484':
return "2.4", "14"
try:
freq = float(freq)
except ValueError:
return "?", "?"
if 2412 <= freq <= 2472:
return "2.4", str(int((freq - 2407) / 5))
elif 3657.5 <= freq <= 3692.5:
return "3", str(int((freq - 3000) / 5))
elif 4915 <= freq <= 4980:
return "5", str(int((freq - 4000) / 5))
elif 5035 <= freq <= 5885:
return "5", str(int((freq - 5000) / 5))
elif 6455 <= freq <= 7115:
return "6", str(int((freq - 5950) / 5))
else:
return "?", "?"
class SdbusNm:
def __init__(self):
self.system_bus = sd_bus_open_system() # We need system bus
set_default_bus(self.system_bus)
self.nm = NetworkManager()
self._callbacks = {
"popup": [],
}
if self.get_wireless_interfaces():
self.wlan_device = self.get_wireless_interfaces()[0]
self.wifi = True
else:
self.wlan_device = None
self.wifi = False
def is_wifi_enabled(self):
return self.nm.wireless_enabled
def get_interfaces(self):
return [NetworkDeviceGeneric(device).interface for device in self.nm.get_devices()]
def get_wireless_interfaces(self):
devices = {path: NetworkDeviceGeneric(path) for path in self.nm.get_devices()}
return [
NetworkDeviceWireless(path)
for path, device in devices.items()
if device.device_type == enums.DeviceType.WIFI
]
def get_primary_interface(self):
if self.nm.primary_connection == '/':
# Nothing connected
if self.wlan_device:
return self.wlan_device.interface
if len(self.get_interfaces()) > 1:
# skips the loopback device
return self.get_interfaces()[1]
return None
gateway = ActiveConnection(self.nm.primary_connection).devices[0]
return NetworkDeviceGeneric(gateway).interface
@staticmethod
def get_known_networks():
known_networks = []
saved_network_paths = NetworkManagerSettings().list_connections()
for netpath in saved_network_paths:
saved_con = NetworkConnectionSettings(netpath)
con_settings = saved_con.get_settings()
# 'type': ('s', '802-11-wireless')
if con_settings['connection']['type'][1] == "802-11-wireless":
known_networks.append({
'SSID': con_settings['802-11-wireless']['ssid'][1].decode(),
'UUID': con_settings['connection']['uuid'][1]
})
return known_networks
def is_known(self, ssid):
return any(net['SSID'] == ssid for net in self.get_known_networks())
def get_ip_address(self):
active_connection_path = self.nm.primary_connection
if not active_connection_path or active_connection_path == '/':
return "?"
active_connection = ActiveConnection(active_connection_path)
ip_info = IPv4Config(active_connection.ip4_config)
return ip_info.address_data[0]['address'][1]
def get_networks(self):
networks = []
if self.wlan_device:
all_aps = [AccessPoint(result) for result in self.wlan_device.access_points]
networks.extend(
{
"SSID": ap.ssid.decode("utf-8"),
"known": self.is_known(ap.ssid.decode("utf-8")),
"security": get_encryption(ap.rsn_flags),
"frequency": WifiChannels(ap.frequency)[0],
"channel": WifiChannels(ap.frequency)[1],
"signal_level": ap.strength,
"max_bitrate": ap.max_bitrate,
"BSSID": ap.hw_address,
}
for ap in all_aps
if ap.ssid
)
return sorted(networks, key=lambda i: i['signal_level'], reverse=True)
return networks
def get_bssid_from_ssid(self, ssid):
return next(net['BSSID'] for net in self.get_networks() if ssid == net['SSID'])
def get_connected_ap(self):
if self.wlan_device.active_access_point == "/":
return None
return AccessPoint(self.wlan_device.active_access_point)
def get_connected_bssid(self):
return self.get_connected_ap().hw_address if self.get_connected_ap() is not None else None
def add_network(self, ssid, psk):
existing_network = NetworkManagerSettings().get_connections_by_id(ssid)
if existing_network:
for network in existing_network:
self.delete_connection_path(network)
properties: NetworkManagerConnectionProperties = {
"connection": {
"id": ("s", ssid),
"uuid": ("s", str(uuid4())),
"type": ("s", "802-11-wireless"),
"interface-name": ("s", self.wlan_device.interface)
},
"802-11-wireless": {
"mode": ("s", "infrastructure"),
"security": ("s", "802-11-wireless-security"),
"ssid": ("ay", ssid.encode("utf-8")),
},
"802-11-wireless-security": {
"key-mgmt": ("s", "wpa-psk"),
"auth-alg": ("s", "open"),
"psk": ("s", psk),
},
"ipv4": {"method": ("s", "auto")},
"ipv6": {"method": ("s", "auto")},
}
try:
msg = f"{ssid}\n" + _("Starting WiFi Association")
self.callback("popup", msg, 1)
NetworkManagerSettings().add_connection(properties)
return True
except Exception as e:
logging.error(f"{e}")
self.callback("popup", f"Error: {e}")
return False
def disconnect_network(self):
self.wlan_device.disconnect()
def delete_network(self, ssid):
connection = NetworkManagerSettings().get_connections_by_id(ssid)
for path in connection:
self.delete_connection_path(path)
@staticmethod
def delete_connection_path(path):
NetworkConnectionSettings(path).delete()
def rescan(self):
return self.wlan_device.request_scan({})
def connect(self, ssid):
connection = NetworkManagerSettings().get_connections_by_id(ssid)
if not connection:
self.callback("popup", f"{ssid} {connection}")
return
msg = f"{ssid}:\n" + _("Starting WiFi Association")
self.callback("popup", msg, 1)
self.nm.activate_connection(connection[0])
def add_callback(self, name, callback):
if name in self._callbacks and callback not in self._callbacks[name]:
self._callbacks[name].append(callback)
def callback(self, cb_type, *args):
if cb_type in self._callbacks:
for cb in self._callbacks[cb_type]:
GLib.idle_add(cb, *args)
def toggle_wifi(self, enable):
self.nm.wireless_enabled = enable

View File

@ -1,347 +0,0 @@
import os
import logging
import re
import socket
import threading
from threading import Thread
from queue import Queue
import gi
gi.require_version("Gtk", "3.0")
from gi.repository import GLib
class WifiManager:
networks_in_supplicant = []
connected = False
_stop_loop = False
def __init__(self, interface, *args, **kwargs):
super().__init__(*args, **kwargs)
self._callbacks = {
"connected": [],
"connecting_status": [],
"scan_results": [],
"popup": [],
}
self._stop_loop = False
self.connected = False
self.connected_ssid = None
self.event = threading.Event()
self.initialized = False
self.interface = interface
self.networks = {}
self.supplicant_networks = {}
self.queue = Queue()
self.timeout = None
ks_socket_file = "/tmp/.KS_wpa_supplicant"
if os.path.exists(ks_socket_file):
os.remove(ks_socket_file)
try:
self.soc = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
self.soc.bind(ks_socket_file)
self.soc.connect(f"/var/run/wpa_supplicant/{interface}")
except Exception as e:
logging.critical(e, exc_info=True)
logging.error(f"Error connecting to wifi socket: {interface}")
return
self.wpa_thread = WpaSocket(self, self.queue, self.callback)
self.wpa_thread.start()
self.initialized = True
self.wpa_cli("ATTACH", False)
self.wpa_cli("SCAN", False)
GLib.idle_add(self.read_wpa_supplicant)
self.timeout = GLib.timeout_add_seconds(180, self.rescan)
def add_callback(self, name, callback):
if name in self._callbacks and callback not in self._callbacks[name]:
self._callbacks[name].append(callback)
def add_network(self, ssid, psk):
for netid in list(self.supplicant_networks):
if self.supplicant_networks[netid]['ssid'] == ssid:
# Modify network
return
# TODO: Add wpa_cli error checking
network_id = self.wpa_cli("ADD_NETWORK")
commands = [
f'ENABLE_NETWORK {network_id}',
'SET_NETWORK %s ssid "%s"' % (network_id, ssid.replace('"', '\"')),
'SET_NETWORK %s psk "%s"' % (network_id, psk.replace('"', '\"'))
]
self.wpa_cli_batch(commands)
self.read_wpa_supplicant()
netid = None
for i in list(self.supplicant_networks):
if self.supplicant_networks[i]['ssid'] == ssid:
netid = i
break
if netid is None:
logging.info("Error adding network")
return False
self.save_wpa_conf()
return True
def callback(self, cb_type, msg):
if cb_type in self._callbacks:
for cb in self._callbacks[cb_type]:
GLib.idle_add(cb, msg)
def connect(self, ssid):
netid = None
for nid, net in self.supplicant_networks.items():
if net['ssid'] == ssid:
netid = nid
break
if netid is None:
logging.info("Wifi network is not defined in wpa_supplicant")
return False
logging.info(f"Attempting to connect to wifi: {netid}")
self.callback("connecting_status", f"Attempting to connect to {ssid}")
self.wpa_cli(f"SELECT_NETWORK {netid}")
self.save_wpa_conf()
def delete_network(self, ssid):
netid = None
for i in list(self.supplicant_networks):
if self.supplicant_networks[i]['ssid'] == ssid:
netid = i
break
if netid is None:
logging.debug("Unable to find network in wpa_supplicant")
return
self.wpa_cli(f"REMOVE_NETWORK {netid}")
for netid in list(self.supplicant_networks):
if self.supplicant_networks[netid]['ssid'] == ssid:
del self.supplicant_networks[netid]
break
self.save_wpa_conf()
def get_connected_ssid(self):
return self.connected_ssid
def get_current_wifi(self):
con_ssid = os.popen("sudo iwgetid -r").read().strip()
con_bssid = os.popen("sudo iwgetid -r -a").read().strip()
# wpa_cli status output is unstable use it as backup only
status = self.wpa_cli("STATUS").split('\n')
variables = {}
for line in status:
arr = line.split('=')
variables[arr[0]] = "=".join(arr[1:])
prev_ssid = self.connected_ssid
if con_ssid != "":
self.connected = True
self.connected_ssid = con_ssid
for ssid, val in self.networks.items():
self.networks[ssid]['connected'] = ssid == con_ssid
if prev_ssid != self.connected_ssid:
for cb in self._callbacks['connected']:
args = self.connected_ssid, prev_ssid
GLib.idle_add(cb, *args)
return [con_ssid, con_bssid]
elif "ssid" in variables and "bssid" in variables:
self.connected = True
self.connected_ssid = variables['ssid']
for ssid, val in self.networks.items():
self.networks[ssid]['connected'] = ssid == variables['ssid']
if prev_ssid != self.connected_ssid:
for cb in self._callbacks['connected']:
args = self.connected_ssid, prev_ssid
GLib.idle_add(cb, *args)
return [variables['ssid'], variables['bssid']]
else:
logging.info("Resetting connected_ssid")
self.connected = False
self.connected_ssid = None
for ssid, val in self.networks.items():
self.networks[ssid]['connected'] = False
if prev_ssid != self.connected_ssid:
for cb in self._callbacks['connected']:
args = self.connected_ssid, prev_ssid
GLib.idle_add(cb, *args)
return None
def get_current_wifi_idle_add(self):
self.get_current_wifi()
return False
def get_network_info(self, ssid=None, mac=None):
if ssid is not None and ssid in self.networks:
return self.networks[ssid]
if mac is not None and ssid is None:
for net in self.networks:
if mac == net['mac']:
return net
return {}
def get_networks(self):
return list(self.networks)
def get_supplicant_networks(self):
return self.supplicant_networks
def read_wpa_supplicant(self):
results = self.wpa_cli("LIST_NETWORKS").split('\n')
results.pop(0)
self.supplicant_networks = {}
self.networks_in_supplicant = []
for net in [n.split('\t') for n in results]:
self.supplicant_networks[net[0]] = {
"ssid": net[1],
"bssid": net[2],
"flags": net[3] if len(net) == 4 else ""
}
self.networks_in_supplicant.append(self.supplicant_networks[net[0]])
def rescan(self):
self.wpa_cli("SCAN", False)
def save_wpa_conf(self):
logging.info("Saving WPA config")
self.wpa_cli("SAVE_CONFIG")
def scan_results(self):
new_networks = []
deleted_networks = list(self.networks)
results = self.wpa_cli("SCAN_RESULTS").split('\n')
results.pop(0)
aps = []
for res in results:
match = re.match("^([a-f0-9:]+)\\s+([0-9]+)\\s+([\\-0-9]+)\\s+(\\S+)\\s+(.+)?", res)
if match:
net = {
"mac": match[1],
"channel": WifiChannels.lookup(match[2])[1],
"connected": False,
"configured": False,
"frequency": match[2],
"flags": match[4],
"signal_level_dBm": match[3],
"ssid": match[5]
}
if "WPA2" in net['flags']:
net['encryption'] = "WPA2"
elif "WPA" in net['flags']:
net['encryption'] = "WPA"
elif "WEP" in net['flags']:
net['encryption'] = "WEP"
else:
net['encryption'] = "off"
aps.append(net)
cur_info = self.get_current_wifi()
self.networks = {}
for ap in aps:
self.networks[ap['ssid']] = ap
if cur_info is not None and cur_info[0] == ap['ssid'] and cur_info[1].lower() == ap['mac'].lower():
self.networks[ap['ssid']]['connected'] = True
for net in list(self.networks):
if net in deleted_networks:
deleted_networks.remove(net)
else:
new_networks.append(net)
if new_networks or deleted_networks:
for cb in self._callbacks['scan_results']:
args = new_networks, deleted_networks
GLib.idle_add(cb, *args)
def wpa_cli(self, command, wait=True):
if wait is False:
self.wpa_thread.skip_command()
self.soc.send(command.encode())
if wait is True:
return self.queue.get()
def wpa_cli_batch(self, commands):
for cmd in commands:
self.wpa_cli(cmd)
class WpaSocket(Thread):
def __init__(self, wm, queue, callback):
super().__init__()
self.queue = queue
self.callback = callback
self.soc = wm.soc
self._stop_loop = False
self.skip_commands = 0
self.wm = wm
def run(self):
logging.debug("Setting up wifi event loop")
while self._stop_loop is False:
try:
msg = self.soc.recv(4096).decode().strip()
except Exception as e:
logging.critical(e, exc_info=True)
# TODO: Socket error
continue
if msg.startswith("<"):
if "CTRL-EVENT-SCAN-RESULTS" in msg:
GLib.idle_add(self.wm.scan_results)
elif "CTRL-EVENT-DISCONNECTED" in msg:
self.callback("connecting_status", msg)
match = re.match('<3>CTRL-EVENT-DISCONNECTED bssid=(\\S+) reason=3 locally_generated=1', msg)
if match:
for net in self.wm.networks:
if self.wm.networks[net]['mac'] == match[1]:
self.wm.networks[net]['connected'] = False
break
elif "Trying to associate" in msg or "CTRL-EVENT-REGDOM-CHANGE" in msg:
self.callback("connecting_status", msg)
elif "CTRL-EVENT-CONNECTED" in msg:
GLib.idle_add(self.wm.get_current_wifi_idle_add)
self.callback("connecting_status", msg)
elif self.skip_commands > 0:
self.skip_commands = self.skip_commands - 1
else:
self.queue.put(msg)
logging.info("Wifi event loop ended")
def skip_command(self):
self.skip_commands = self.skip_commands + 1
class WifiChannels:
@staticmethod
def lookup(freq: str):
if freq == '2484':
return "2.4", "14"
try:
freq = float(freq)
except ValueError:
return None
if 2412 <= freq <= 2472:
return "2.4", str(int((freq - 2407) / 5))
elif 3657.5 <= freq <= 3692.5:
return "3", str(int((freq - 3000) / 5))
elif 4915 <= freq <= 4980:
return "5", str(int((freq - 4000) / 5))
elif 5035 <= freq <= 5885:
return "5", str(int((freq - 5000) / 5))
elif 6455 <= freq <= 7115:
return "6", str(int((freq - 5950) / 5))
else:
return "?", "?"

View File

@ -1,272 +0,0 @@
# Network in KlipperScreen is a connection in NetworkManager
# Interface in KlipperScreen is a device in NetworkManager
import logging
import uuid
import dbus
import gi
gi.require_version('Gdk', '3.0')
from gi.repository import GLib
from contextlib import suppress
from ks_includes.wifi import WifiChannels
from ks_includes import NetworkManager
from dbus.mainloop.glib import DBusGMainLoop
class WifiManager:
networks_in_supplicant = []
def __init__(self, interface_name, *args, **kwargs):
super().__init__(*args, **kwargs)
DBusGMainLoop(set_as_default=True)
self._callbacks = {
"connected": [],
"connecting_status": [],
"scan_results": [],
"popup": [],
}
self.connected = False
self.connected_ssid = None
self.interface_name = interface_name
self.known_networks = {} # List of known connections
self.visible_networks = {} # List of visible access points
self.ssid_by_path = {}
self.path_by_ssid = {}
self.hidden_ssid_index = 0
self.wifi_dev = NetworkManager.NetworkManager.GetDeviceByIpIface(interface_name)
self.wifi_dev.OnAccessPointAdded(self._ap_added)
self.wifi_dev.OnAccessPointRemoved(self._ap_removed)
self.wifi_dev.OnStateChanged(self._ap_state_changed)
for ap in self.wifi_dev.GetAccessPoints():
self._add_ap(ap)
self._update_known_connections()
self.initialized = True
def _update_known_connections(self):
self.known_networks = {}
for con in NetworkManager.Settings.ListConnections():
settings = con.GetSettings()
if "802-11-wireless" in settings:
ssid = settings["802-11-wireless"]['ssid']
self.known_networks[ssid] = con
def _ap_added(self, nm, interface, signal, access_point):
with suppress(NetworkManager.ObjectVanished):
access_point.OnPropertiesChanged(self._ap_prop_changed)
ssid = self._add_ap(access_point)
for cb in self._callbacks['scan_results']:
args = (cb, [ssid], [])
GLib.idle_add(*args)
def _ap_removed(self, dev, interface, signal, access_point):
path = access_point.object_path
if path in self.ssid_by_path:
ssid = self.ssid_by_path[path]
self._remove_ap(path)
for cb in self._callbacks['scan_results']:
args = (cb, [ssid], [])
GLib.idle_add(*args)
def _ap_state_changed(self, nm, interface, signal, old_state, new_state, reason):
msg = ""
if new_state in (NetworkManager.NM_DEVICE_STATE_UNKNOWN, NetworkManager.NM_DEVICE_STATE_REASON_UNKNOWN):
msg = "State is unknown"
elif new_state == NetworkManager.NM_DEVICE_STATE_UNMANAGED:
msg = "Error: Not managed by NetworkManager"
elif new_state == NetworkManager.NM_DEVICE_STATE_UNAVAILABLE:
msg = "Error: Not available for use:\nReasons may include the wireless switched off, missing firmware, etc."
elif new_state == NetworkManager.NM_DEVICE_STATE_DISCONNECTED:
msg = "Currently disconnected"
elif new_state == NetworkManager.NM_DEVICE_STATE_PREPARE:
msg = "Preparing the connection to the network"
elif new_state == NetworkManager.NM_DEVICE_STATE_CONFIG:
msg = "Connecting to the requested network..."
elif new_state == NetworkManager.NM_DEVICE_STATE_NEED_AUTH:
msg = "Authorizing"
elif new_state == NetworkManager.NM_DEVICE_STATE_IP_CONFIG:
msg = "Requesting IP addresses and routing information"
elif new_state == NetworkManager.NM_DEVICE_STATE_IP_CHECK:
msg = "Checking whether further action is required for the requested network connection"
elif new_state == NetworkManager.NM_DEVICE_STATE_SECONDARIES:
msg = "Waiting for a secondary connection (like a VPN)"
elif new_state == NetworkManager.NM_DEVICE_STATE_ACTIVATED:
msg = "Connected"
elif new_state == NetworkManager.NM_DEVICE_STATE_DEACTIVATING:
msg = "A disconnection from the current network connection was requested"
elif new_state == NetworkManager.NM_DEVICE_STATE_FAILED:
msg = "Failed to connect to the requested network"
self.callback("popup", msg)
elif new_state == NetworkManager.NM_DEVICE_STATE_REASON_DEPENDENCY_FAILED:
msg = "A dependency of the connection failed"
elif new_state == NetworkManager.NM_DEVICE_STATE_REASON_CARRIER:
msg = ""
else:
logging.info(f"State {new_state}")
if msg != "":
self.callback("connecting_status", msg)
if new_state == NetworkManager.NM_DEVICE_STATE_ACTIVATED:
self.connected = True
for cb in self._callbacks['connected']:
args = (cb, self.get_connected_ssid(), None)
GLib.idle_add(*args)
else:
self.connected = False
def _ap_prop_changed(self, ap, interface, signal, properties):
pass
def _add_ap(self, ap):
ssid = ap.Ssid
if ssid == "":
ssid = _("Hidden") + f" {self.hidden_ssid_index}"
self.hidden_ssid_index += 1
self.ssid_by_path[ap.object_path] = ssid
self.path_by_ssid[ssid] = ap.object_path
self.visible_networks[ap.object_path] = ap
return ssid
def _remove_ap(self, path):
self.ssid_by_path.pop(path, None)
self.visible_networks.pop(path, None)
def add_callback(self, name, callback):
if name in self._callbacks and callback not in self._callbacks[name]:
self._callbacks[name].append(callback)
def callback(self, cb_type, msg):
if cb_type in self._callbacks:
for cb in self._callbacks[cb_type]:
GLib.idle_add(cb, msg)
def add_network(self, ssid, psk):
aps = self._visible_networks_by_ssid()
if ssid in aps:
ap = aps[ssid]
new_connection = {
'802-11-wireless': {
'mode': 'infrastructure',
'security': '802-11-wireless-security',
'ssid': ssid
},
'802-11-wireless-security': {
'auth-alg': 'open',
'key-mgmt': 'wpa-psk',
'psk': psk
},
'connection': {
'id': ssid,
'type': '802-11-wireless',
'uuid': str(uuid.uuid4())
},
'ipv4': {
'method': 'auto'
},
'ipv6': {
'method': 'auto'
}
}
try:
NetworkManager.Settings.AddConnection(new_connection)
except dbus.exceptions.DBusException as e:
msg = _("Invalid password") if "802-11-wireless-security.psk" in e else f"{e}"
self.callback("popup", msg)
logging.info(f"Error adding network {e}")
self._update_known_connections()
return True
def connect(self, ssid):
if ssid in self.known_networks:
conn = self.known_networks[ssid]
with suppress(NetworkManager.ObjectVanished):
msg = f"Connecting to: {ssid}"
logging.info(msg)
self.callback("connecting_status", msg)
NetworkManager.NetworkManager.ActivateConnection(conn, self.wifi_dev, "/")
def delete_network(self, ssid):
for ssid in self.known_networks:
con = self.known_networks[ssid]
if con.GetSettings()['connection']['id'] == ssid:
con.Delete()
self._update_known_connections()
def get_connected_ssid(self):
if self.wifi_dev.SpecificDevice().ActiveAccessPoint:
return self.wifi_dev.SpecificDevice().ActiveAccessPoint.Ssid
return None
def _get_connected_ap(self):
return self.wifi_dev.SpecificDevice().ActiveAccessPoint
def _visible_networks_by_ssid(self):
aps = self.wifi_dev.GetAccessPoints()
ret = {}
for ap in aps:
with suppress(NetworkManager.ObjectVanished):
ret[ap.Ssid] = ap
return ret
def get_network_info(self, ssid):
netinfo = {}
if ssid in self.known_networks:
con = self.known_networks[ssid]
with suppress(NetworkManager.ObjectVanished):
settings = con.GetSettings()
if settings and '802-11-wireless' in settings:
netinfo.update({
"ssid": settings['802-11-wireless']['ssid'],
"connected": self.get_connected_ssid() == ssid
})
path = self.path_by_ssid[ssid]
aps = self.visible_networks
if path in aps:
ap = aps[path]
with suppress(NetworkManager.ObjectVanished):
netinfo.update({
"mac": ap.HwAddress,
"channel": WifiChannels.lookup(ap.Frequency)[1],
"configured": ssid in self.known_networks,
"frequency": str(ap.Frequency),
"flags": ap.Flags,
"ssid": ssid,
"connected": self._get_connected_ap() == ap,
"encryption": self._get_encryption(ap.RsnFlags),
"signal_level_dBm": str(ap.Strength)
})
return netinfo
@staticmethod
def _get_encryption(flags):
encryption = ""
if (flags & NetworkManager.NM_802_11_AP_SEC_PAIR_WEP40 or
flags & NetworkManager.NM_802_11_AP_SEC_PAIR_WEP104 or
flags & NetworkManager.NM_802_11_AP_SEC_GROUP_WEP40 or
flags & NetworkManager.NM_802_11_AP_SEC_GROUP_WEP104):
encryption += "WEP "
if (flags & NetworkManager.NM_802_11_AP_SEC_PAIR_TKIP or
flags & NetworkManager.NM_802_11_AP_SEC_GROUP_TKIP):
encryption += "TKIP "
if (flags & NetworkManager.NM_802_11_AP_SEC_PAIR_CCMP or
flags & NetworkManager.NM_802_11_AP_SEC_GROUP_CCMP):
encryption += "AES "
if flags & NetworkManager.NM_802_11_AP_SEC_KEY_MGMT_PSK:
encryption += "WPA-PSK "
if flags & NetworkManager.NM_802_11_AP_SEC_KEY_MGMT_802_1X:
encryption += "802.1x "
return encryption.strip()
def get_networks(self):
return list(set(list(self.known_networks.keys()) + list(self.ssid_by_path.values())))
def get_supplicant_networks(self):
return {ssid: {"ssid": ssid} for ssid in self.known_networks.keys()}
def rescan(self):
try:
self.wifi_dev.RequestScan({})
except dbus.exceptions.DBusException as e:
logging.error(f"Error during rescan {e}")

View File

@ -1,143 +1,99 @@
import logging
import os
import gi
import netifaces
gi.require_version("Gtk", "3.0")
from gi.repository import Gtk, GLib, Pango
from ks_includes.screen_panel import ScreenPanel
from ks_includes.sdbus_nm import SdbusNm
class Panel(ScreenPanel):
initialized = False
def __init__(self, screen, title):
super().__init__(screen, title)
self.show_add = False
self.networks = {}
self.interface = None
self.prev_network = None
self.update_timeout = None
self.network_interfaces = netifaces.interfaces()
self.wireless_interfaces = [iface for iface in self.network_interfaces if iface.startswith('wl')]
self.wifi = None
self.use_network_manager = os.system('systemctl is-active --quiet NetworkManager.service') == 0
if self.wireless_interfaces:
logging.info(f"Found wireless interfaces: {self.wireless_interfaces}")
if self.use_network_manager:
logging.info("Using NetworkManager")
from ks_includes.wifi_nm import WifiManager
else:
logging.info("Using wpa_cli")
from ks_includes.wifi import WifiManager
self.wifi = WifiManager(self.wireless_interfaces[0])
else:
logging.info(_("No wireless interface has been found"))
self.network_list = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, hexpand=True, vexpand=True)
self.network_rows = {}
self.networks = {}
self.sdbus_nm = SdbusNm()
self.wifi_signal_icons = {
'excellent': self._gtk.PixbufFromIcon('wifi_excellent'),
'good': self._gtk.PixbufFromIcon('wifi_good'),
'fair': self._gtk.PixbufFromIcon('wifi_fair'),
'weak': self._gtk.PixbufFromIcon('wifi_weak'),
}
# Get IP Address
gws = netifaces.gateways()
if "default" in gws and netifaces.AF_INET in gws["default"]:
self.interface = gws["default"][netifaces.AF_INET][1]
else:
ints = netifaces.interfaces()
if 'lo' in ints:
ints.pop(ints.index('lo'))
self.interface = ints[0] if len(ints) > 0 else 'lo'
self.network_interfaces = self.sdbus_nm.get_interfaces()
logging.info(f"Network interfaces: {self.network_interfaces}")
self.labels['networks'] = {}
self.wireless_interfaces = [iface.interface for iface in self.sdbus_nm.get_wireless_interfaces()]
logging.info(f"Wireless interfaces: {self.wireless_interfaces}")
self.interface = self.sdbus_nm.get_primary_interface()
logging.info(f"Primary interface: {self.interface}")
self.labels['interface'] = Gtk.Label(hexpand=True)
self.labels['interface'].set_text(_("Interface") + f': {self.interface} ')
self.labels['ip'] = Gtk.Label(hexpand=True)
ifadd = netifaces.ifaddresses(self.interface)
if ifadd.get(netifaces.AF_INET):
self.labels['ip'].set_text(f"IP: {ifadd[netifaces.AF_INET][0]['addr']} ")
if self.interface is not None:
self.labels['interface'].set_text(_("Interface") + f': {self.interface}')
self.labels['ip'].set_text(f"IP: {self.sdbus_nm.get_ip_address()}")
reload_networks = self._gtk.Button("refresh", None, "color1", self.bts)
reload_networks.connect("clicked", self.reload_networks)
reload_networks.set_hexpand(False)
self.reload_button = self._gtk.Button("refresh", None, "color1", self.bts)
self.reload_button.set_no_show_all(True)
self.reload_button.show()
self.reload_button.connect("clicked", self.reload_networks)
self.reload_button.set_hexpand(False)
self.wifi_toggle = Gtk.Switch(
width_request=round(self._gtk.font_size * 2),
height_request=round(self._gtk.font_size),
active=self.sdbus_nm.is_wifi_enabled()
)
self.wifi_toggle.connect("notify::active", self.toggle_wifi)
sbox = Gtk.Box(hexpand=True, vexpand=False)
sbox.add(self.labels['interface'])
sbox.add(self.labels['ip'])
sbox.add(reload_networks)
sbox.add(self.reload_button)
sbox.add(self.wifi_toggle)
scroll = self._gtk.ScrolledWindow()
self.labels['main_box'] = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, vexpand=True)
box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, vexpand=True)
self.labels['networklist'] = Gtk.Grid()
if self.wifi is not None and self.wifi.initialized:
box.pack_start(sbox, False, False, 5)
box.pack_start(scroll, True, True, 0)
if self.sdbus_nm.wifi:
self.labels['main_box'].pack_start(sbox, False, False, 5)
GLib.idle_add(self.load_networks)
scroll.add(self.labels['networklist'])
scroll.add(self.network_list)
self.wifi.add_callback("connected", self.connected_callback)
self.wifi.add_callback("scan_results", self.scan_callback)
self.wifi.add_callback("popup", self.popup_callback)
if self.update_timeout is None:
self.update_timeout = GLib.timeout_add_seconds(5, self.update_all_networks)
self.sdbus_nm.add_callback("popup", self.popup_callback)
else:
self._screen.show_popup_message(_("No wireless interface has been found"), level=2)
self.labels['networkinfo'] = Gtk.Label()
self.labels['networkinfo'].get_style_context().add_class('temperature_entry')
box.pack_start(self.labels['networkinfo'], False, False, 0)
scroll.add(self.labels['networkinfo'])
self.update_single_network_info()
if self.update_timeout is None:
self.update_timeout = GLib.timeout_add_seconds(5, self.update_single_network_info)
self.content.add(box)
self.labels['main_box'] = box
self.initialized = True
self.labels['main_box'].pack_start(scroll, True, True, 0)
self.content.add(self.labels['main_box'])
def load_networks(self, widget=None):
networks = self.wifi.get_networks()
if not networks:
return
for net in networks:
self.add_network(net, False)
self.update_all_networks()
if widget:
GLib.timeout_add_seconds(10, self._gtk.Button_busy, widget, False)
def popup_callback(self, msg, level=3):
self._screen.show_popup_message(msg, level)
def load_networks(self):
for net in self.sdbus_nm.get_networks():
self.add_network(net['BSSID'])
GLib.timeout_add_seconds(10, self._gtk.Button_busy, self.reload_button, False)
self.content.show_all()
return False
def add_network(self, ssid, show=True):
if ssid is None:
return
ssid = ssid.strip()
if ssid in list(self.networks):
def add_network(self, bssid):
if bssid in self.network_rows:
logging.info(f"{bssid} already in list")
return
configured_networks = self.wifi.get_supplicant_networks()
network_id = -1
for net in list(configured_networks):
if configured_networks[net]['ssid'] == ssid:
network_id = net
display_name = _("Hidden") if ssid.startswith("\x00") else f"{ssid}"
netinfo = self.wifi.get_network_info(ssid)
connected_ssid = self.wifi.get_connected_ssid()
if netinfo is None:
logging.debug("Couldn't get netinfo")
netinfo = {'connected': connected_ssid == ssid}
name = Gtk.Label(hexpand=True, halign=Gtk.Align.START, wrap=True, wrap_mode=Pango.WrapMode.WORD_CHAR)
if connected_ssid == ssid:
display_name += " (" + _("Connected") + ")"
name.set_markup(f"<big><b>{display_name}</b></big>")
else:
name.set_label(display_name)
info = Gtk.Label(halign=Gtk.Align.START)
labels = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, vexpand=True,
halign=Gtk.Align.START, valign=Gtk.Align.CENTER)
labels.add(name)
labels.add(info)
net = next(net for net in self.sdbus_nm.get_networks() if bssid == net['BSSID'])
ssid = net['SSID']
connect = self._gtk.Button("load", None, "color3", self.bts)
connect.connect("clicked", self.connect_network, ssid)
@ -145,54 +101,79 @@ class Panel(ScreenPanel):
connect.set_halign(Gtk.Align.END)
delete = self._gtk.Button("delete", None, "color3", self.bts)
delete.connect("clicked", self.remove_wifi_network, ssid)
delete.connect("clicked", self.remove_confirm_dialog, ssid, bssid)
delete.set_hexpand(False)
delete.set_halign(Gtk.Align.END)
network = Gtk.Box(spacing=5, hexpand=True, vexpand=False)
network.get_style_context().add_class("frame-item")
network.add(labels)
buttons = Gtk.Box(spacing=5)
if network_id != -1 or netinfo['connected']:
buttons.pack_end(connect, False, False, 0)
buttons.pack_end(delete, False, False, 0)
else:
buttons.pack_end(connect, False, False, 0)
network.add(buttons)
self.networks[ssid] = network
nets = sorted(list(self.networks), reverse=False)
if connected_ssid in nets:
nets.remove(connected_ssid)
nets.insert(0, connected_ssid)
if nets.index(ssid) is not None:
pos = nets.index(ssid)
name = Gtk.Label(hexpand=True, halign=Gtk.Align.START, wrap=True, wrap_mode=Pango.WrapMode.WORD_CHAR)
if bssid == self.sdbus_nm.get_connected_bssid():
ssid += ' (' + _("Connected") + ')'
name.set_markup(f"<big><b>{ssid}</b></big>")
else:
logging.info("Error: SSID not in nets")
return
name.set_markup(f"<b>{ssid}</b>")
if net['known']:
buttons.add(delete)
buttons.add(connect)
self.labels['networks'][ssid] = {
info = Gtk.Label(halign=Gtk.Align.START)
labels = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, vexpand=True,
halign=Gtk.Align.START, valign=Gtk.Align.CENTER)
labels.add(name)
labels.add(info)
icon = self._gtk.Image('wifi_weak')
self.network_rows[bssid] = Gtk.Box(spacing=5, hexpand=True, vexpand=False)
self.network_rows[bssid].get_style_context().add_class("frame-item")
self.network_rows[bssid].add(icon)
self.network_rows[bssid].add(labels)
self.network_rows[bssid].add(buttons)
self.networks[bssid] = {
"connect": connect,
"delete": delete,
"icon": icon,
"info": info,
"name": name,
"row": network
"row": self.network_rows[bssid],
}
self.labels['networklist'].insert_row(pos)
self.labels['networklist'].attach(self.networks[ssid], 0, pos, 1, 1)
if show:
self.labels['networklist'].show()
self.network_list.add(self.network_rows[bssid])
def remove_confirm_dialog(self, widget, ssid, bssid):
label = Gtk.Label(wrap=True, vexpand=True)
label.set_markup(_("Do you want to forget or disconnect %s?") % ssid)
buttons = [
{"name": _("Forget"), "response": Gtk.ResponseType.OK, "style": 'dialog-warning'},
{"name": _("Cancel"), "response": Gtk.ResponseType.CANCEL, "style": 'dialog-error'},
]
if bssid == self.sdbus_nm.get_connected_bssid():
buttons.insert(0, {"name": _("Disconnect"), "response": Gtk.ResponseType.APPLY, "style": 'dialog-info'})
self._gtk.Dialog(_("Remove network"), buttons, label, self.confirm_removal, ssid)
def confirm_removal(self, dialog, response_id, ssid):
self._gtk.remove_dialog(dialog)
if response_id == Gtk.ResponseType.CANCEL:
return
bssid = self.sdbus_nm.get_bssid_from_ssid(ssid)
self.remove_network_from_list(bssid)
if response_id == Gtk.ResponseType.OK:
logging.info(f"Deleting {ssid}")
self.sdbus_nm.delete_network(ssid)
if response_id == Gtk.ResponseType.APPLY:
logging.info(f"Disconnecting {ssid}")
self.sdbus_nm.disconnect_network()
def add_new_network(self, widget, ssid):
self._screen.remove_keyboard()
result = self.wifi.add_network(ssid, self.labels['network_psk'].get_text())
result = self.sdbus_nm.add_network(ssid, self.labels['network_psk'].get_text())
self.close_add_network()
if result:
self.connect_network(widget, ssid, False)
self.connect_network(widget, ssid, showadd=False)
else:
self._screen.show_popup_message(f"Error adding network {ssid}")
self._screen.show_popup_message(_("Invalid password"))
def back(self):
if self.show_add:
@ -200,16 +181,6 @@ class Panel(ScreenPanel):
return True
return False
def check_missing_networks(self):
networks = self.wifi.get_networks()
for net in list(self.networks):
if net in networks:
networks.remove(net)
for net in networks:
self.add_network(net, False)
self.labels['networklist'].show_all()
def close_add_network(self):
if not self.show_add:
return
@ -223,71 +194,27 @@ class Panel(ScreenPanel):
del self.labels[i]
self.show_add = False
def popup_callback(self, msg):
self._screen.show_popup_message(msg)
def connected_callback(self, ssid, prev_ssid):
logging.info("Now connected to a new network")
if ssid is not None:
self.remove_network(ssid)
if prev_ssid is not None:
self.remove_network(prev_ssid)
self.check_missing_networks()
def connect_network(self, widget, ssid, showadd=True):
isdef = any(net['ssid'] == ssid for netid, net in self.wifi.get_supplicant_networks().items())
if not isdef:
if showadd:
self.show_add_network(widget, ssid)
self.deactivate()
if showadd and not self.sdbus_nm.is_known(ssid):
self.show_add_network(widget, ssid)
self.activate()
return
self.prev_network = self.wifi.get_connected_ssid()
bssid = self.sdbus_nm.get_bssid_from_ssid(ssid)
if bssid and bssid in self.network_rows:
self.remove_network_from_list(bssid)
self.sdbus_nm.connect(ssid)
self.update_all_networks()
self.activate()
buttons = [
{"name": _("Close"), "response": Gtk.ResponseType.CANCEL}
]
scroll = self._gtk.ScrolledWindow()
self.labels['connecting_info'] = Gtk.Label(
label=_("Starting WiFi Association"), halign=Gtk.Align.START, valign=Gtk.Align.START, wrap=True)
scroll.add(self.labels['connecting_info'])
self._gtk.Dialog(_("Starting WiFi Association"), buttons, scroll, self._gtk.remove_dialog)
self._screen.show_all()
if ssid in list(self.networks):
self.remove_network(ssid)
if self.prev_network in list(self.networks):
self.remove_network(self.prev_network)
self.wifi.add_callback("connecting_status", self.connecting_status_callback)
self.wifi.connect(ssid)
def connecting_status_callback(self, msg):
self.labels['connecting_info'].set_text(f"{self.labels['connecting_info'].get_text()}\n{msg}")
self.labels['connecting_info'].show_all()
def remove_network(self, ssid, show=True):
if ssid not in list(self.networks):
def remove_network_from_list(self, bssid):
if bssid not in self.network_rows:
logging.error(f"{bssid} not in rows")
return
for i in range(len(self.labels['networklist'])):
if self.networks[ssid] == self.labels['networklist'].get_child_at(0, i):
self.labels['networklist'].remove_row(i)
self.labels['networklist'].show()
del self.networks[ssid]
del self.labels['networks'][ssid]
return
def remove_wifi_network(self, widget, ssid):
self.wifi.delete_network(ssid)
self.remove_network(ssid)
self.check_missing_networks()
def scan_callback(self, new_networks, old_networks):
for net in old_networks:
self.remove_network(net, False)
for net in new_networks:
self.add_network(net, False)
self.content.show_all()
self.network_list.remove(self.network_rows[bssid])
del self.network_rows[bssid]
del self.networks[bssid]
return
def show_add_network(self, widget, ssid):
if self.show_add:
@ -323,100 +250,96 @@ class Panel(ScreenPanel):
self.show_add = True
def update_all_networks(self):
for network in list(self.networks):
self.update_network_info(network)
self.interface = self.sdbus_nm.get_primary_interface()
self.labels['interface'].set_text(_("Interface") + f': {self.interface}')
self.labels['ip'].set_text(f"IP: {self.sdbus_nm.get_ip_address()}")
nets = self.sdbus_nm.get_networks()
remove = [bssid for bssid in self.network_rows.keys() if bssid not in [net['BSSID'] for net in nets]]
for bssid in remove:
self.remove_network_from_list(bssid)
for net in nets:
if net['BSSID'] not in self.network_rows.keys():
self.add_network(net['BSSID'])
self.update_network_info(net)
for i, net in enumerate(nets):
for child in self.network_list.get_children():
if child == self.network_rows[net['BSSID']]:
self.network_list.reorder_child(child, i)
self.network_list.show_all()
return True
def update_network_info(self, ssid):
info = freq = encr = chan = lvl = ipv4 = ipv6 = ""
if ssid not in list(self.networks) or ssid not in self.labels['networks']:
logging.info(f"Unknown SSID {ssid}")
def update_network_info(self, net):
if net['BSSID'] not in self.network_rows.keys() or net['BSSID'] not in self.networks:
logging.info(f"Unknown SSID {net['SSID']}")
return
netinfo = self.wifi.get_network_info(ssid)
if netinfo.get('connected') or self.wifi.get_connected_ssid() == ssid:
ifadd = netifaces.ifaddresses(self.interface)
if ifadd.get(netifaces.AF_INET):
ipv4 = f"<b>IPv4:</b> {ifadd[netifaces.AF_INET][0]['addr']}"
self.labels['ip'].set_text(f"IP: {ifadd[netifaces.AF_INET][0]['addr']} ")
if ifadd.get(netifaces.AF_INET6):
ipv6 = f"<b>IPv6:</b> {ifadd[netifaces.AF_INET6][0]['addr'].split('%')[0]}"
info = _("Password saved") + '\n' if net['known'] else ""
chan = _("Channel") + f' {net["channel"]}'
max_bitrate = _("Max:") + f"{self.format_speed(net['max_bitrate'])}"
self.networks[net['BSSID']]['icon'].set_from_pixbuf(self.get_signal_strength_icon(net["signal_level"]))
self.networks[net['BSSID']]['info'].set_markup(
"<small>"
f"{info}"
f"{net['security']}\n"
f"{max_bitrate}\n"
f"{net['frequency']} Ghz {chan} {net['signal_level']} %\n"
f"{net['BSSID']}"
"</small>"
)
info = '<b>' + _("Hostname") + f':</b> {os.uname().nodename}\n{ipv4}\n{ipv6}'
else:
self.labels['networks'][ssid]['name'].set_label(_("Hidden") if ssid.startswith("\x00") else f"{ssid}")
if "psk" in netinfo:
info = _("Password saved")
if "encryption" in netinfo and netinfo['encryption'] != "off":
encr = netinfo['encryption'].upper()
if "frequency" in netinfo:
freq = "2.4 GHz" if netinfo['frequency'][:1] == "2" else "5 Ghz"
if "channel" in netinfo:
chan = _("Channel") + f' {netinfo["channel"]}'
if "signal_level_dBm" in netinfo:
unit = "%" if self.use_network_manager else _("dBm")
lvl = f"{netinfo['signal_level_dBm']} {unit}"
icon = self.signal_strength(int(netinfo["signal_level_dBm"]))
if 'icon' not in self.labels['networks'][ssid]:
self.labels['networks'][ssid]['row'].add(icon)
self.labels['networks'][ssid]['row'].reorder_child(icon, 0)
self.labels['networks'][ssid]['icon'] = icon
self.labels['networks'][ssid]['icon'] = icon
self.labels['networks'][ssid]['info'].set_markup(f"{info}\n<small>{encr} {freq} {chan} {lvl}</small>")
self.labels['networks'][ssid]['row'].show_all()
def signal_strength(self, signal_level):
def get_signal_strength_icon(self, signal_level):
# networkmanager uses percentage not dbm
# the bars of nmcli are aligned near this breakpoints
exc = 77 if self.use_network_manager else -50
good = 60 if self.use_network_manager else -60
fair = 35 if self.use_network_manager else -70
if signal_level > exc:
return self._gtk.Image('wifi_excellent')
elif signal_level > good:
return self._gtk.Image('wifi_good')
elif signal_level > fair:
return self._gtk.Image('wifi_fair')
if signal_level > 75:
return self.wifi_signal_icons['excellent']
elif signal_level > 60:
return self.wifi_signal_icons['good']
elif signal_level > 30:
return self.wifi_signal_icons['fair']
else:
return self._gtk.Image('wifi_weak')
return self.wifi_signal_icons['weak']
def update_single_network_info(self):
ifadd = netifaces.ifaddresses(self.interface)
ipv6 = f"{ifadd[netifaces.AF_INET6][0]['addr'].split('%')[0]}" if ifadd.get(netifaces.AF_INET6) else ""
if netifaces.AF_INET in ifadd and ifadd[netifaces.AF_INET]:
ipv4 = f"{ifadd[netifaces.AF_INET][0]['addr']} "
self.labels['ip'].set_text(f"IP: {ifadd[netifaces.AF_INET][0]['addr']} ")
else:
ipv4 = ""
self.labels['networkinfo'].set_markup(
f'<b>{self.interface}</b>\n\n'
+ '<b>' + _("Hostname") + f':</b> {os.uname().nodename}\n'
f'<b>IPv4:</b> {ipv4}\n'
f'<b>IPv6:</b> {ipv6}'
f'<b>IPv4:</b> {self.sdbus_nm.get_ip_address()}\n'
)
self.labels['networkinfo'].show_all()
return True
def reload_networks(self, widget=None):
self.networks = {}
self.labels['networklist'].remove_column(0)
if self.wifi is not None and self.wifi.initialized:
self.deactivate()
del self.network_rows
self.network_rows = {}
for child in self.network_list.get_children():
self.network_list.remove(child)
if self.sdbus_nm is not None and self.sdbus_nm.wifi:
if widget:
self._gtk.Button_busy(widget, True)
self.wifi.rescan()
GLib.idle_add(self.load_networks, widget)
self.sdbus_nm.rescan()
self.load_networks()
self.activate()
def activate(self):
if self.initialized:
self.reload_networks()
if self.update_timeout is None:
if self.wifi is not None and self.wifi.initialized:
self.update_timeout = GLib.timeout_add_seconds(5, self.update_all_networks)
else:
self.update_timeout = GLib.timeout_add_seconds(5, self.update_single_network_info)
if self.update_timeout is None:
if self.sdbus_nm is not None and self.sdbus_nm.wifi:
if self.reload_button.get_sensitive():
self._gtk.Button_busy(self.reload_button, True)
self.sdbus_nm.rescan()
self.load_networks()
self.update_timeout = GLib.timeout_add_seconds(5, self.update_all_networks)
else:
self.update_timeout = GLib.timeout_add_seconds(5, self.update_single_network_info)
def deactivate(self):
if self.update_timeout is not None:
GLib.source_remove(self.update_timeout)
self.update_timeout = None
def toggle_wifi(self, switch, gparams):
enable = switch.get_active()
if enable:
self.reload_button.show()
else:
self.reload_button.hide()
logging.info(f"WiFi {enable}")
self.sdbus_nm.toggle_wifi(enable)

View File

@ -1,8 +1,7 @@
jinja2==3.1.4;python_version>="3.6"
requests==2.31.0;python_version>="3.7"
netifaces==0.11.0
six==1.16.0
dbus-python==1.3.2
sdbus==0.11.1;python_version>="3.8"
sdbus_networkmanager==2.0.0;python_version>="3.8"
# libmpv-dev 0.33 is required for 1.0
python-mpv==0.5.2;python_version<"3.10"