Huge updates: ControlCenter, DropDown Menu
This commit is contained in:
34 files changed
+3261
-1440
No files matched your search
+186
-4
@@ -1,19 +1,81 @@
|
||||
# type: ignore
|
||||
from loguru import logger
|
||||
from fabric.core.service import Service, Signal, Property
|
||||
from fabric.notifications import Notification
|
||||
from typing import List
|
||||
|
||||
|
||||
class ModusService(Service):
|
||||
@Signal
|
||||
def bluetooth_changed(self, new_bluetooth: str) -> None: ...
|
||||
|
||||
@Signal
|
||||
def volume_changed(self, new_volume: int) -> None: ...
|
||||
|
||||
@Signal
|
||||
def wlan_changed(self, new_wlan: str) -> None: ...
|
||||
|
||||
@Signal
|
||||
def battery_changed(self, new_battery: str) -> None: ...
|
||||
|
||||
@Signal
|
||||
def dock_apps_changed(self, new_dock_apps: str) -> None: ...
|
||||
|
||||
@Signal
|
||||
def clear_all_changed(self, value: bool) -> None: ...
|
||||
|
||||
@Signal
|
||||
def notification_count_changed(self, value: int) -> None: ...
|
||||
|
||||
@Signal
|
||||
def dont_disturb_changed(self, value: bool) -> None: ...
|
||||
|
||||
@Signal
|
||||
def current_active_app_name_changed(self, value: str) -> None: ...
|
||||
|
||||
@Signal
|
||||
def music_changed(self, value: str) -> None: ...
|
||||
|
||||
@Signal
|
||||
def current_dropdown_changed(self, value: str) -> None: ...
|
||||
|
||||
@Signal
|
||||
def dropdowns_hide_changed(self, value: bool) -> None: ...
|
||||
|
||||
@Signal
|
||||
def dock_width_changed(self, value: int) -> None: ...
|
||||
|
||||
@Signal
|
||||
def dock_height_changed(self, value: int) -> None: ...
|
||||
|
||||
@Signal
|
||||
def dock_hidden_changed(self, value: bool) -> None: ...
|
||||
|
||||
@Signal
|
||||
def show_notificationcenter_changed(self, value: bool) -> None: ...
|
||||
@Property(bool, flags="read-write", default_value=False)
|
||||
def show_notificationcenter(self) -> bool:
|
||||
return self._show_notificationcenter
|
||||
|
||||
@Property(str, flags="read-write")
|
||||
def current_active_app_name(self) -> str:
|
||||
return self._current_active_app_name
|
||||
|
||||
@Property(str, flags="read-write")
|
||||
def bluetooth(self) -> str:
|
||||
return self._bluetooth
|
||||
|
||||
@Property(str, flags="read-write")
|
||||
def wlan(self) -> str:
|
||||
return self._wlan
|
||||
|
||||
@Property(str, flags="read-write")
|
||||
def battery(self) -> str:
|
||||
return self._battery
|
||||
|
||||
@Property(int, flags="read-write")
|
||||
def volume(self) -> int:
|
||||
return self._volume
|
||||
|
||||
@Property(str, flags="read-write")
|
||||
def dock_apps(self) -> str:
|
||||
return self._dock_apps
|
||||
|
||||
@Property(str, flags="read-write")
|
||||
def notification_count(self) -> int:
|
||||
@@ -23,12 +85,112 @@ class ModusService(Service):
|
||||
def dont_disturb(self) -> bool:
|
||||
return self._dont_disturb
|
||||
|
||||
@Property(str, flags="read-write")
|
||||
def music(self) -> str:
|
||||
return self._music
|
||||
|
||||
@Property(str, flags="read-write")
|
||||
def current_dropdown(self) -> str:
|
||||
return self._current_dropdown
|
||||
|
||||
@Property(bool, flags="read-write", default_value=False)
|
||||
def dropdowns_hide(self) -> bool:
|
||||
return self._dropdowns_hide
|
||||
|
||||
@Property(int, flags="read-write")
|
||||
def dock_width(self) -> int:
|
||||
return self._dock_width
|
||||
|
||||
@Property(int, flags="read-write")
|
||||
def dock_height(self) -> int:
|
||||
return self._dock_height
|
||||
|
||||
@Property(bool, flags="read-write", default_value=False)
|
||||
def dock_hidden(self) -> bool:
|
||||
return self._dock_hidden
|
||||
|
||||
@Property(bool, flags="read-write", default_value=False)
|
||||
def show_notificationcenter(self) -> bool:
|
||||
return self._show_notificationcenter
|
||||
|
||||
@current_active_app_name.setter
|
||||
def current_active_app_name(self, value: str):
|
||||
if value != self._current_active_app_name:
|
||||
self._current_active_app_name = value
|
||||
self.current_active_app_name_changed(value)
|
||||
|
||||
@volume.setter
|
||||
def volume(self, value: int):
|
||||
if value != self._volume:
|
||||
self._name = value
|
||||
self.volume_changed(value)
|
||||
|
||||
@wlan.setter
|
||||
def wlan(self, value: str):
|
||||
if value != self._wlan:
|
||||
self._wlan = value
|
||||
self.wlan_changed(value)
|
||||
|
||||
@battery.setter
|
||||
def battery(self, value: str):
|
||||
if value != self._battery:
|
||||
self._battery = value
|
||||
self.battery_changed(value)
|
||||
|
||||
@bluetooth.setter
|
||||
def bluetooth(self, value: str):
|
||||
if value != self._bluetooth:
|
||||
self._bluetooth = value
|
||||
self.bluetooth_changed(value)
|
||||
|
||||
@dock_apps.setter
|
||||
def dock_apps(self, value: str):
|
||||
if value != self._dock_apps:
|
||||
self._dock_apps = value
|
||||
self.dock_apps_changed(value)
|
||||
|
||||
@dont_disturb.setter
|
||||
def dont_disturb(self, value: bool):
|
||||
if value != self._dont_disturb:
|
||||
self._dont_disturb = value
|
||||
self.dont_disturb_changed(value)
|
||||
|
||||
@music.setter
|
||||
def music(self, value: str):
|
||||
if value != self._music:
|
||||
self._music = value
|
||||
self.music_changed(value)
|
||||
|
||||
@current_dropdown.setter
|
||||
def current_dropdown(self, value: str):
|
||||
if value != self._current_dropdown:
|
||||
self._current_dropdown = value
|
||||
self.current_dropdown_changed(value)
|
||||
|
||||
@dropdowns_hide.setter
|
||||
def dropdowns_hide(self, value: bool):
|
||||
if value != self._dropdowns_hide:
|
||||
self._dropdowns_hide = value
|
||||
self.dropdowns_hide_changed(value)
|
||||
|
||||
@dock_width.setter
|
||||
def dock_width(self, value: int):
|
||||
if value != self._dock_width:
|
||||
self._dock_width = value
|
||||
self.dock_width_changed(value)
|
||||
|
||||
@dock_height.setter
|
||||
def dock_height(self, value: int):
|
||||
if value != self._dock_height:
|
||||
self._dock_height = value
|
||||
self.dock_height_changed(value)
|
||||
|
||||
@dock_hidden.setter
|
||||
def dock_hidden(self, value: bool):
|
||||
if value != self._dock_hidden:
|
||||
self._dock_hidden = value
|
||||
self.dock_hidden_changed(value)
|
||||
|
||||
@show_notificationcenter.setter
|
||||
def show_notificationcenter(self, value: bool):
|
||||
if value != self._show_notificationcenter:
|
||||
@@ -41,11 +203,24 @@ class ModusService(Service):
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self._volume = 0
|
||||
self._wlan = ""
|
||||
self._battery = ""
|
||||
self._bluetooth = ""
|
||||
self._dock_apps = ""
|
||||
self._notifications = []
|
||||
self._notification_count = len(self._notifications)
|
||||
self._dont_disturb = False
|
||||
self._current_active_app_name = ""
|
||||
self._music = ""
|
||||
self._current_dropdown = None
|
||||
self._dropdowns_hide = False
|
||||
self._dock_hidden = False
|
||||
self._show_notificationcenter = False
|
||||
|
||||
self._dock_width = 0
|
||||
self._dock_height = 0
|
||||
|
||||
self.notifications = []
|
||||
|
||||
def remove_notification(self, id: int):
|
||||
@@ -83,3 +258,10 @@ class ModusService(Service):
|
||||
Notification.deserialize(data) for data in self._notifications
|
||||
]
|
||||
return self.notifications
|
||||
|
||||
|
||||
global modus_service
|
||||
try:
|
||||
modus_service = ModusService()
|
||||
except Exception as e:
|
||||
logger.error("[Main] Failed to create EnvShellService:", e)
|
||||
+364
-561
@@ -1,301 +1,17 @@
|
||||
from typing import Optional
|
||||
import subprocess
|
||||
from typing import Any, List, Literal
|
||||
|
||||
import gi
|
||||
from fabric.core.service import Property, Service, Signal
|
||||
from fabric.utils import bulk_connect, get_enum_member_name, snake_case_to_kebab_case
|
||||
from gi.repository import NM, GLib
|
||||
from fabric.utils import bulk_connect
|
||||
from gi.repository import Gio
|
||||
from loguru import logger
|
||||
|
||||
gi.require_version("NM", "1.0") # Ensure the correct version is loaded
|
||||
|
||||
|
||||
class NetworkClient(Service):
|
||||
"""A service to manage network devices"""
|
||||
|
||||
@Signal
|
||||
def wifi_device_added(self) -> None: ...
|
||||
|
||||
@Signal
|
||||
def ethernet_device_added(self) -> None: ...
|
||||
|
||||
@Signal
|
||||
def wifi_device_removed(self) -> None: ...
|
||||
|
||||
@Signal
|
||||
def ethernet_device_removed(self) -> None: ...
|
||||
|
||||
@Signal
|
||||
def changed(self) -> None: ...
|
||||
|
||||
@Signal
|
||||
def ready(self) -> None: ...
|
||||
|
||||
@Property(bool, "readable", default_value=False)
|
||||
def is_ready(self) -> bool:
|
||||
"""Returns whether the client is ready to use."""
|
||||
return self._client is not None
|
||||
|
||||
@Property(list, "readable")
|
||||
def connections(self) -> Optional[list]:
|
||||
"""Returns the active connections, if available."""
|
||||
return self._client.get_property("connections") if self._client else None
|
||||
|
||||
@Property(object, "readable")
|
||||
def wifi_device(self) -> Optional[object]:
|
||||
"""Returns the WiFi device if available."""
|
||||
return self._wifi_device
|
||||
|
||||
@Property(object, "readable")
|
||||
def ethernet_device(self) -> Optional[object]:
|
||||
"""Returns the Ethernet device if available."""
|
||||
return self._ethernet_device
|
||||
|
||||
@Property(str, "readable")
|
||||
def primary_connection(self) -> Optional[str]:
|
||||
"""Returns the primary connection if available."""
|
||||
return self._client.get_property("primary_connection") if self._client else None
|
||||
|
||||
@Property(str, "readable")
|
||||
def active_connection(self) -> Optional[str]:
|
||||
"""Returns the active connection if available."""
|
||||
return self._client.get_property("active_connection") if self._client else None
|
||||
|
||||
@Property(str, "readable")
|
||||
def state(self) -> str:
|
||||
"""Returns the current network state."""
|
||||
if not self._client:
|
||||
return "disconnected"
|
||||
return snake_case_to_kebab_case(
|
||||
get_enum_member_name(
|
||||
self._client.get_property("state"), default="disconnected"
|
||||
)
|
||||
)
|
||||
|
||||
@Property(str, "readable")
|
||||
def connectivity(self) -> str:
|
||||
"""Returns the connectivity state."""
|
||||
if not self._client:
|
||||
return "disconnected"
|
||||
return snake_case_to_kebab_case(
|
||||
get_enum_member_name(
|
||||
self._client.get_property("connectivity"), default="disconnected"
|
||||
)
|
||||
)
|
||||
|
||||
@Property(list, "readable")
|
||||
def devices(self) -> Optional[list]:
|
||||
"""Returns the list of network devices if available."""
|
||||
return self._client.get_property("devices") if self._client else None
|
||||
|
||||
@Property(str, "readable")
|
||||
def hostname(self) -> Optional[str]:
|
||||
"""Returns the hostname if available."""
|
||||
return self._client.get_property("hostname") if self._client else None
|
||||
|
||||
@Property(bool, "read-write", default_value=False)
|
||||
def networking_enabled(self) -> bool:
|
||||
"""Checks if networking is enabled."""
|
||||
return (
|
||||
self._client.get_property("networking_enabled") if self._client else False
|
||||
)
|
||||
|
||||
@networking_enabled.setter
|
||||
def networking_enabled(self, value: bool):
|
||||
"""Sets the networking state if the client is available."""
|
||||
if self._client:
|
||||
self._client.set_property("networking_enabled", value)
|
||||
|
||||
@Property(bool, "read-write", default_value=False)
|
||||
def wireless_enabled(self) -> bool:
|
||||
"""Checks if wireless networking is enabled."""
|
||||
return self._client.get_property("wireless_enabled") if self._client else False
|
||||
|
||||
@wireless_enabled.setter
|
||||
def wireless_enabled(self, value: bool):
|
||||
"""Sets the wireless networking state if the client is available."""
|
||||
if self._client:
|
||||
self._client.set_property("wireless_enabled", value)
|
||||
|
||||
def __init__(self, **kwargs):
|
||||
super().__init__(**kwargs)
|
||||
|
||||
self._client: NM.Client = None
|
||||
self._wifi_device: Wifi | None = None
|
||||
self._ethernet_device: Ethernet | None = None
|
||||
|
||||
logger.info("[Network] Initializing client asynchronously...")
|
||||
|
||||
# Start async NM client initialization
|
||||
NM.Client.new_async(None, self.on_client_ready)
|
||||
|
||||
def on_client_ready(self, source, result):
|
||||
"""Callback when NM.Client is ready."""
|
||||
try:
|
||||
self._client = NM.Client.new_finish(result) # Retrieve client instance
|
||||
logger.info("[Network] NM.Client initialized successfully!")
|
||||
|
||||
# Connect signals
|
||||
bulk_connect(
|
||||
self._client,
|
||||
{
|
||||
"device-added": lambda _, device: self.on_device_added(
|
||||
device=device
|
||||
),
|
||||
"device-removed": lambda _, device: self.on_device_removed(
|
||||
device=device
|
||||
),
|
||||
"notify::state": lambda *args: self.notifier("state"),
|
||||
"notify::networking-enabled": lambda *args: self.notifier(
|
||||
"networking-enabled"
|
||||
),
|
||||
"notify::wireless-enabled": lambda *args: self.notifier(
|
||||
"wireless-enabled"
|
||||
),
|
||||
# "notify::primary-connection": lambda *args: self.notifier('primary-connection'),
|
||||
# "notify::active-connection": lambda *args: self.notifier('active-connection'),
|
||||
# "active-connection-added": lambda *args: self.emit("changed"),
|
||||
# "active-connection-removed": lambda *args: self.emit("changed")
|
||||
},
|
||||
)
|
||||
|
||||
# Process devices AFTER client is ready
|
||||
for device in self.do_get_raw_devices():
|
||||
self.on_device_added(device=device)
|
||||
|
||||
self.notify("state")
|
||||
self.notify("networking-enabled")
|
||||
self.notify("wireless-enabled")
|
||||
self.notify("is-ready")
|
||||
self.ready.emit()
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"[Network] Error initializing NM.Client: {e}")
|
||||
|
||||
def do_get_raw_devices(self) -> list[NM.Device]:
|
||||
return [
|
||||
dev
|
||||
for dev in self.devices
|
||||
if dev.get_device_type() in (NM.DeviceType.WIFI, NM.DeviceType.ETHERNET)
|
||||
]
|
||||
|
||||
def on_device_added(self, device):
|
||||
device_type = device.get_device_type()
|
||||
if device_type == NM.DeviceType.WIFI and not self._wifi_device:
|
||||
logger.info("[Network] WiFi device detected, initializing...")
|
||||
self._wifi_device = Wifi(client=self, device=device)
|
||||
self.wifi_device_added.emit()
|
||||
|
||||
elif device_type == NM.DeviceType.ETHERNET and not self._ethernet_device:
|
||||
logger.info("[Network] Ethernet device detected, initializing...")
|
||||
self._ethernet_device = Ethernet(client=self, device=device)
|
||||
self.ethernet_device_added.emit()
|
||||
|
||||
def on_device_removed(self, device):
|
||||
if device == self._wifi_device:
|
||||
logger.info("[Network] WiFi device removed.")
|
||||
self._wifi_device = None
|
||||
self.wifi_device_removed.emit()
|
||||
|
||||
elif device == self._ethernet_device:
|
||||
logger.info("[Network] Ethernet device removed.")
|
||||
self._ethernet_device = None
|
||||
self.ethernet_device_removed.emit()
|
||||
|
||||
def toggle_network(self):
|
||||
"""Enable or disable Network"""
|
||||
self.networking_enabled = not self.networking_enabled
|
||||
|
||||
def deactivate_connection(self, connection):
|
||||
"""Disconnect"""
|
||||
self._client.deactivate_connection_async(connection, None, None)
|
||||
|
||||
def notifier(self, name):
|
||||
self.notify(name)
|
||||
self.emit("changed")
|
||||
|
||||
|
||||
class AccessPoint(Service):
|
||||
"""A service to manage access points"""
|
||||
|
||||
@Signal
|
||||
def changed(self) -> None: ...
|
||||
|
||||
@Property(object, "readable")
|
||||
def device(self) -> object:
|
||||
return self._device
|
||||
|
||||
@Property(int, "readable")
|
||||
def strength(self) -> int:
|
||||
return self._ap.get_property("strength")
|
||||
|
||||
@Property(int, "readable")
|
||||
def frequency(self) -> int:
|
||||
return self._ap.get_property("frequency")
|
||||
|
||||
@Property(str, "readable")
|
||||
def bssid(self) -> str:
|
||||
return self._ap.get_property("bssid") if self._ap else None
|
||||
|
||||
@Property(str, "readable")
|
||||
def hw_address(self) -> str:
|
||||
return self._ap.get_property("hw_address")
|
||||
|
||||
@Property(str, "readable")
|
||||
def ssid(self) -> str:
|
||||
ssid = self._ap.get_ssid()
|
||||
return NM.utils_ssid_to_utf8(ssid.get_data()) if ssid else "Unknown"
|
||||
|
||||
@Property(str, "readable")
|
||||
def icon(self) -> str:
|
||||
return {
|
||||
80: "network-wireless-signal-excellent-symbolic",
|
||||
60: "network-wireless-signal-good-symbolic",
|
||||
40: "network-wireless-signal-ok-symbolic",
|
||||
20: "network-wireless-signal-weak-symbolic",
|
||||
00: "network-wireless-signal-none-symbolic",
|
||||
}.get(
|
||||
min(80, 20 * round(self.strength / 20)),
|
||||
"network-wireless-no-route-symbolic",
|
||||
)
|
||||
|
||||
@Property(bool, "readable", default_value=False)
|
||||
def requires_password(self) -> bool:
|
||||
ssid = self.ssid
|
||||
settings = self._client.connections
|
||||
connection = None
|
||||
for setting in settings:
|
||||
wifi_setting = setting.get_setting_wireless()
|
||||
if (
|
||||
wifi_setting
|
||||
and NM.utils_ssid_to_utf8(wifi_setting.get_ssid().get_data()) == ssid
|
||||
):
|
||||
connection = setting
|
||||
break
|
||||
if not connection:
|
||||
return bool(self._ap.get_wpa_flags() or self._ap.get_rsn_flags())
|
||||
return False
|
||||
|
||||
@Property(bool, "readable", default_value=False)
|
||||
def is_active(self) -> bool:
|
||||
if self._device.active_access_point:
|
||||
return self.bssid == self._device.active_access_point.get_bssid()
|
||||
return False
|
||||
|
||||
def __init__(self, device: "Wifi", ap: NM.AccessPoint, **kwargs):
|
||||
super().__init__(**kwargs)
|
||||
self._client: NetworkClient = device.client
|
||||
self._device: Wifi = device
|
||||
self._ap: NM.AccessPoint = ap
|
||||
|
||||
self._ap.connect("notify::strength", lambda *args: self.notifier("strength"))
|
||||
self._device.connect(
|
||||
"notify::active-access-point", lambda *args: self.notifier("is-active")
|
||||
)
|
||||
|
||||
def notifier(self, name: str, *args):
|
||||
self.notify(name)
|
||||
self.emit("changed")
|
||||
return
|
||||
try:
|
||||
gi.require_version("NM", "1.0")
|
||||
from gi.repository import NM
|
||||
except ValueError:
|
||||
raise logger.error("Failed to start network manager")
|
||||
|
||||
|
||||
class Wifi(Service):
|
||||
@@ -305,244 +21,286 @@ class Wifi(Service):
|
||||
def changed(self) -> None: ...
|
||||
|
||||
@Signal
|
||||
def ap_added(self, ap: AccessPoint) -> None: ...
|
||||
def enabled(self) -> bool: ...
|
||||
|
||||
@Signal
|
||||
def ap_removed(self, ap: AccessPoint) -> None: ...
|
||||
def scanning(self, is_scanning: bool) -> None: ...
|
||||
|
||||
@Property(NetworkClient, "readable")
|
||||
def client(self) -> NetworkClient:
|
||||
"""Returns the client"""
|
||||
return self._client
|
||||
|
||||
@Property(bool, "read-write", default_value=False)
|
||||
def wireless_enabled(self) -> bool:
|
||||
"""Returns if the wifi is enabled"""
|
||||
return self._client.get_property("wireless_enabled")
|
||||
|
||||
@wireless_enabled.setter
|
||||
def wireless_enabled(self, value: bool):
|
||||
return self._client.set_property("wireless_enabled", value)
|
||||
|
||||
@Property(list[AccessPoint], "readable")
|
||||
def access_points(self) -> list[AccessPoint]:
|
||||
return sorted(
|
||||
self._access_points.values(), key=lambda x: x.is_active, reverse=True
|
||||
)
|
||||
|
||||
@Property(AccessPoint, "readable")
|
||||
def active_access_point(self) -> Optional[AccessPoint]:
|
||||
return self._active_access_point
|
||||
|
||||
def __init__(self, client: NetworkClient, device: NM.DeviceWifi, **kwargs):
|
||||
super().__init__(**kwargs)
|
||||
self._client: NetworkClient = client
|
||||
def __init__(self, client: NM.Client, device: NM.DeviceWifi, **kwargs):
|
||||
self._client: NM.Client = client
|
||||
self._device: NM.DeviceWifi = device
|
||||
self._active_access_point: NM.AccessPoint | None = None
|
||||
self._access_points: dict[str, AccessPoint] = {}
|
||||
|
||||
bulk_connect(
|
||||
self._device,
|
||||
{
|
||||
"notify::active-access-point": lambda *args: self.on_access_point_activated(),
|
||||
"access-point-added": lambda _, ap: self.on_access_point_added(ap=ap),
|
||||
"access-point-removed": lambda _, ap: self.on_access_point_removed(
|
||||
ap=ap
|
||||
),
|
||||
# "state-changed": lambda device, new, old, reason: self.on_state_changed(new),
|
||||
},
|
||||
)
|
||||
self._ap: NM.AccessPoint | None = None
|
||||
self._ap_signal: int | None = None
|
||||
super().__init__(**kwargs)
|
||||
|
||||
self._client.connect(
|
||||
"notify::wireless-enabled", lambda *args: self.notifier("wireless-enabled")
|
||||
"notify::wireless-enabled",
|
||||
lambda *args: self.notifier("enabled", args),
|
||||
)
|
||||
if self._device:
|
||||
bulk_connect(
|
||||
self._device,
|
||||
{
|
||||
"notify::active-access-point": self._activate_ap,
|
||||
"access-point-added": lambda *_: self.emit("changed"),
|
||||
"access-point-removed": lambda *_: self.emit("changed"),
|
||||
"state-changed": self.ap_update,
|
||||
},
|
||||
)
|
||||
self._activate_ap()
|
||||
|
||||
for ap in self.do_get_access_points():
|
||||
self.on_access_point_added(ap=ap)
|
||||
|
||||
self.on_access_point_activated()
|
||||
|
||||
def on_state_changed(self, state):
|
||||
def ap_update(self, *_):
|
||||
self.emit("changed")
|
||||
for sn in [
|
||||
"enabled",
|
||||
"internet",
|
||||
"strength",
|
||||
"frequency",
|
||||
"access-points",
|
||||
"ssid",
|
||||
"state",
|
||||
"icon-name",
|
||||
]:
|
||||
self.notify(sn)
|
||||
|
||||
def do_get_access_points(self):
|
||||
return self._device.get_access_points()
|
||||
def _activate_ap(self, *_):
|
||||
if self._ap:
|
||||
self._ap.disconnect(self._ap_signal)
|
||||
self._ap = self._device.get_active_access_point()
|
||||
if not self._ap:
|
||||
return
|
||||
|
||||
def on_access_point_added(self, ap):
|
||||
ssid = ap.get_ssid()
|
||||
ssid = NM.utils_ssid_to_utf8(ssid.get_data()) if ssid else "Unknown"
|
||||
|
||||
access_point: AccessPoint = AccessPoint(ap=ap, device=self)
|
||||
|
||||
self._access_points[ssid] = access_point
|
||||
|
||||
self.ap_added.emit(access_point)
|
||||
|
||||
# self.notifier("access-points")
|
||||
|
||||
logger.info(f"[Wifi] New access point added with ssid: {ssid}")
|
||||
|
||||
def on_access_point_removed(self, ap):
|
||||
ssid = ap.get_ssid()
|
||||
ssid = NM.utils_ssid_to_utf8(ssid.get_data()) if ssid else "Unknown"
|
||||
|
||||
if not (access_point := self._access_points.pop(ssid, None)):
|
||||
return logger.warning(
|
||||
f"[Network] tried to remove a unknwon access point with ssid {ssid}"
|
||||
)
|
||||
|
||||
self.ap_removed.emit(access_point)
|
||||
|
||||
logger.info(f"[Wifi] Access point with ssid: {ssid} removed.")
|
||||
|
||||
def on_access_point_activated(self):
|
||||
if self._device.get_active_access_point():
|
||||
self._active_access_point: AccessPoint = AccessPoint(
|
||||
ap=self._device.get_active_access_point(), device=self
|
||||
)
|
||||
|
||||
else:
|
||||
self._active_access_point = None
|
||||
|
||||
self.notifier("active-access-point")
|
||||
|
||||
logger.info("[Wifi] New active connection")
|
||||
|
||||
def disconnect_wifi(self):
|
||||
"""Disconnect from the current WiFi network."""
|
||||
active_connection = self._device.get_active_connection()
|
||||
self._client.deactivate_connection(active_connection)
|
||||
logger.info("[Wifi] Wifi network disconnected")
|
||||
|
||||
def scan(self):
|
||||
self._device.request_scan_async(
|
||||
None,
|
||||
lambda device, result: [
|
||||
device.request_scan_finish(result),
|
||||
self.notifier("access-points"),
|
||||
],
|
||||
)
|
||||
logger.info("[Wifi] Scan started")
|
||||
self._ap_signal = self._ap.connect(
|
||||
"notify::strength", lambda *args: self.ap_update()
|
||||
) # type: ignore
|
||||
|
||||
def toggle_wifi(self):
|
||||
"""Enable or disable WiFi"""
|
||||
self.wireless_enabled = not self.wireless_enabled
|
||||
self._client.wireless_set_enabled(not self._client.wireless_get_enabled())
|
||||
|
||||
def connect_to_wifi(self, ap: AccessPoint, password: str = None):
|
||||
"""Connect to a WiFi network."""
|
||||
def get_ap_security(
|
||||
self, nm_ap: NM.AccessPoint
|
||||
) -> Literal["WEP", "WPA1", "WPA2", "802.1X", "unsecured"]:
|
||||
"""Parse the security flags to return a string with 'WPA2', etc."""
|
||||
flags = nm_ap.get_flags()
|
||||
wpa_flags = nm_ap.get_wpa_flags()
|
||||
rsn_flags = nm_ap.get_rsn_flags()
|
||||
sec_str = ""
|
||||
if (
|
||||
(flags & getattr(NM, "80211ApFlags").PRIVACY)
|
||||
and (wpa_flags == 0)
|
||||
and (rsn_flags == 0)
|
||||
):
|
||||
sec_str += " WEP"
|
||||
if wpa_flags != 0:
|
||||
sec_str += " WPA1"
|
||||
if rsn_flags != 0:
|
||||
sec_str += " WPA2"
|
||||
if (wpa_flags & getattr(NM, "80211ApSecurityFlags").KEY_MGMT_802_1X) or (
|
||||
rsn_flags & getattr(NM, "80211ApSecurityFlags").KEY_MGMT_802_1X
|
||||
):
|
||||
sec_str += " 802.1X"
|
||||
|
||||
ssid = ap.ssid
|
||||
# If there is no security use "--"
|
||||
if sec_str == "":
|
||||
sec_str = "unsecured"
|
||||
return sec_str.lstrip()
|
||||
|
||||
if ssid == "Unknown":
|
||||
logger.error("Invalid access point data")
|
||||
return False
|
||||
|
||||
logger.info(f"Connecting to WiFi SSID: {ssid}")
|
||||
|
||||
# Check for existing connections
|
||||
settings = self._client.connections
|
||||
connection = None
|
||||
|
||||
for setting in settings:
|
||||
wifi_setting = setting.get_setting_wireless()
|
||||
if (
|
||||
wifi_setting
|
||||
and NM.utils_ssid_to_utf8(wifi_setting.get_ssid().get_data()) == ssid
|
||||
):
|
||||
connection = setting
|
||||
break
|
||||
|
||||
if not connection:
|
||||
# Create a new connection profile
|
||||
logger.info(f"Creating new WiFi connection for SSID '{ssid}'")
|
||||
connection = NM.SimpleConnection.new()
|
||||
|
||||
# Required connection settings
|
||||
s_con = NM.SettingConnection.new()
|
||||
s_con.set_property(NM.SETTING_CONNECTION_ID, ssid)
|
||||
s_con.set_property(NM.SETTING_CONNECTION_TYPE, "802-11-wireless")
|
||||
s_con.set_property(
|
||||
NM.SETTING_CONNECTION_INTERFACE_NAME, self._device.get_iface()
|
||||
) # Set interface name
|
||||
connection.add_setting(s_con)
|
||||
|
||||
# Wireless settings
|
||||
s_wifi = NM.SettingWireless.new()
|
||||
s_wifi.set_property(NM.SETTING_WIRELESS_SSID, GLib.Bytes.new(ssid.encode()))
|
||||
s_wifi.set_property(
|
||||
NM.SETTING_WIRELESS_MODE, "infrastructure"
|
||||
) # Ensure mode is correct
|
||||
connection.add_setting(s_wifi)
|
||||
|
||||
# Security settings (only if password is required and provided)
|
||||
if ap.requires_password:
|
||||
if not password:
|
||||
logger.error("Password required but not provided")
|
||||
return False
|
||||
|
||||
s_sec = NM.SettingWirelessSecurity.new()
|
||||
s_sec.set_property(NM.SETTING_WIRELESS_SECURITY_KEY_MGMT, "wpa-psk")
|
||||
s_sec.set_property(NM.SETTING_WIRELESS_SECURITY_PSK, password)
|
||||
connection.add_setting(s_sec)
|
||||
|
||||
# IPv4 settings
|
||||
s_ipv4 = NM.SettingIP4Config.new()
|
||||
s_ipv4.set_property("method", "auto")
|
||||
connection.add_setting(s_ipv4)
|
||||
|
||||
# IPv6 settings
|
||||
s_ipv6 = NM.SettingIP6Config.new()
|
||||
s_ipv6.set_property("method", "auto")
|
||||
connection.add_setting(s_ipv6)
|
||||
|
||||
# Callback for async connection
|
||||
def on_connection_added(client, result, ssid, device):
|
||||
try:
|
||||
new_connection = client.add_connection_finish(result)
|
||||
if not new_connection:
|
||||
logger.error(f"Failed to create connection for '{ssid}'")
|
||||
return
|
||||
|
||||
logger.info(f"Connection for '{ssid}' added successfully")
|
||||
|
||||
client.activate_connection_async(
|
||||
new_connection,
|
||||
device,
|
||||
None,
|
||||
None,
|
||||
lambda c, r: (
|
||||
logger.info(f"Connected to '{ssid}'")
|
||||
if c
|
||||
else logger.error(f"Failed to connect to '{ssid}'")
|
||||
),
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to add connection: {e}")
|
||||
|
||||
# Save the new connection
|
||||
self._client._client.add_connection_async(
|
||||
connection,
|
||||
True,
|
||||
def scan(self):
|
||||
"""Start scanning for WiFi networks and emit scanning signal"""
|
||||
if self._device:
|
||||
self.emit("scanning", True) # Emit signal that scanning has started
|
||||
self._device.request_scan_async(
|
||||
None,
|
||||
lambda c, r: on_connection_added(c, r, ssid, self._device),
|
||||
lambda device, result: [
|
||||
device.request_scan_finish(result),
|
||||
self.emit(
|
||||
"scanning", False
|
||||
), # Emit signal that scanning has stopped
|
||||
],
|
||||
)
|
||||
|
||||
# Activate the connection
|
||||
self._client._client.activate_connection_async(
|
||||
connection,
|
||||
self._device,
|
||||
None,
|
||||
None,
|
||||
lambda client, result: logger.info(f"Connected to '{ssid}'"),
|
||||
)
|
||||
def is_active_ap(self, name) -> bool:
|
||||
return self._ap.get_bssid() == name if self._ap else False
|
||||
|
||||
return True
|
||||
|
||||
def notifier(self, name: str, *args):
|
||||
def notifier(self, name: str, *_):
|
||||
self.notify(name)
|
||||
self.emit("changed")
|
||||
return
|
||||
|
||||
def forget_access_point(self, ssid):
|
||||
try:
|
||||
# List all saved connections
|
||||
result = subprocess.check_output(
|
||||
"nmcli connection show", shell=True, text=True
|
||||
)
|
||||
|
||||
# Find connection ID that matches SSID
|
||||
for line in result.splitlines():
|
||||
if ssid in line:
|
||||
connection_id = line.split()[0]
|
||||
subprocess.check_call(
|
||||
f"nmcli connection delete id '{connection_id}'", shell=True
|
||||
)
|
||||
logger.info(
|
||||
f"[NetworkService] Deleted saved connection: {connection_id}"
|
||||
)
|
||||
return True
|
||||
|
||||
logger.warning(
|
||||
f"[NetworkService] No saved connection found for SSID: {ssid}"
|
||||
)
|
||||
return False
|
||||
|
||||
except subprocess.CalledProcessError as e:
|
||||
logger.error(f"[NetworkService] Error forgetting connection: {e}")
|
||||
return False
|
||||
|
||||
def connect_network(
|
||||
self, ssid: str, password: str = "", remember: bool = True
|
||||
) -> bool:
|
||||
"""Connect to a WiFi network"""
|
||||
if not ssid:
|
||||
logger.error("[NetworkService] SSID cannot be empty")
|
||||
return False
|
||||
try:
|
||||
# First try to connect using saved connection
|
||||
try:
|
||||
subprocess.run(["nmcli", "con", "up", ssid], check=True)
|
||||
return True
|
||||
except subprocess.CalledProcessError:
|
||||
# If saved connection fails, try with password if provided
|
||||
if password:
|
||||
cmd = [
|
||||
"nmcli",
|
||||
"device",
|
||||
"wifi",
|
||||
"connect",
|
||||
ssid,
|
||||
"password",
|
||||
password,
|
||||
]
|
||||
if not remember:
|
||||
cmd.extend(["--temporary"])
|
||||
subprocess.run(cmd, check=True)
|
||||
return True
|
||||
return False
|
||||
except subprocess.CalledProcessError as e:
|
||||
logger.error(f"[NetworkService] Failed connecting to network: {e}")
|
||||
return False
|
||||
|
||||
def disconnect_network(self, ssid: str) -> bool:
|
||||
"""Disconnect from a WiFi network"""
|
||||
if not ssid:
|
||||
logger.error("[NetworkService] SSID cannot be empty")
|
||||
return False
|
||||
try:
|
||||
subprocess.run(["nmcli", "con", "down", ssid], check=True)
|
||||
return True
|
||||
except subprocess.CalledProcessError as e:
|
||||
logger.error(f"[NetworkService] Failed disconnecting from network: {e}")
|
||||
return False
|
||||
|
||||
@Property(bool, "read-write", default_value=False)
|
||||
def enabled(self) -> bool: # noqa: F811
|
||||
return bool(self._client.wireless_get_enabled())
|
||||
|
||||
@enabled.setter
|
||||
def enabled(self, value: bool):
|
||||
self._client.wireless_set_enabled(value)
|
||||
|
||||
@Property(int, "readable")
|
||||
def strength(self):
|
||||
return self._ap.get_strength() if self._ap else -1
|
||||
|
||||
@Property(str, "readable")
|
||||
def icon_name(self):
|
||||
if not self._ap:
|
||||
return "network-wireless-disabled-symbolic"
|
||||
|
||||
if self.internet == "activated":
|
||||
return {
|
||||
80: "network-wireless-signal-excellent-symbolic",
|
||||
60: "network-wireless-signal-good-symbolic",
|
||||
40: "network-wireless-signal-ok-symbolic",
|
||||
20: "network-wireless-signal-weak-symbolic",
|
||||
00: "network-wireless-signal-none-symbolic",
|
||||
}.get(
|
||||
min(80, 20 * round(self._ap.get_strength() / 20)),
|
||||
"network-wireless-no-route-symbolic",
|
||||
)
|
||||
if self.internet == "activating":
|
||||
return "network-wireless-acquiring-symbolic"
|
||||
|
||||
return "network-wireless-offline-symbolic"
|
||||
|
||||
@Property(int, "readable")
|
||||
def frequency(self):
|
||||
return self._ap.get_frequency() if self._ap else -1
|
||||
|
||||
@Property(int, "readable")
|
||||
def internet(self):
|
||||
return {
|
||||
NM.ActiveConnectionState.ACTIVATED: "activated",
|
||||
NM.ActiveConnectionState.ACTIVATING: "activating",
|
||||
NM.ActiveConnectionState.DEACTIVATING: "deactivating",
|
||||
NM.ActiveConnectionState.DEACTIVATED: "deactivated",
|
||||
}.get(
|
||||
self._device.get_active_connection().get_state(),
|
||||
"unknown",
|
||||
)
|
||||
|
||||
@Property(object, "readable")
|
||||
def access_points(self) -> List[object]:
|
||||
points: list[NM.AccessPoint] = self._device.get_access_points()
|
||||
|
||||
def make_ap_dict(ap: NM.AccessPoint):
|
||||
return {
|
||||
"bssid": ap.get_bssid(),
|
||||
"last_seen": ap.get_last_seen(),
|
||||
"ssid": NM.utils_ssid_to_utf8(ap.get_ssid().get_data())
|
||||
if ap.get_ssid()
|
||||
else "Unknown",
|
||||
"active-ap": self._ap,
|
||||
"strength": ap.get_strength(),
|
||||
"frequency": ap.get_frequency(),
|
||||
"icon-name": {
|
||||
80: "network-wireless-signal-excellent-symbolic",
|
||||
60: "network-wireless-signal-good-symbolic",
|
||||
40: "network-wireless-signal-ok-symbolic",
|
||||
20: "network-wireless-signal-weak-symbolic",
|
||||
00: "network-wireless-signal-none-symbolic",
|
||||
}.get(
|
||||
min(80, 20 * round(ap.get_strength() / 20)),
|
||||
"network-wireless-no-route-symbolic",
|
||||
),
|
||||
}
|
||||
|
||||
return list(map(make_ap_dict, points))
|
||||
|
||||
@Property(str, "readable")
|
||||
def ssid(self):
|
||||
if not self._ap:
|
||||
return "Disconnected"
|
||||
ssid = self._ap.get_ssid().get_data()
|
||||
return NM.utils_ssid_to_utf8(ssid) if ssid else "Unknown"
|
||||
|
||||
@Property(int, "readable")
|
||||
def state(self):
|
||||
return {
|
||||
NM.DeviceState.UNMANAGED: "unmanaged",
|
||||
NM.DeviceState.UNAVAILABLE: "unavailable",
|
||||
NM.DeviceState.DISCONNECTED: "disconnected",
|
||||
NM.DeviceState.PREPARE: "prepare",
|
||||
NM.DeviceState.CONFIG: "config",
|
||||
NM.DeviceState.NEED_AUTH: "need_auth",
|
||||
NM.DeviceState.IP_CONFIG: "ip_config",
|
||||
NM.DeviceState.IP_CHECK: "ip_check",
|
||||
NM.DeviceState.SECONDARIES: "secondaries",
|
||||
NM.DeviceState.ACTIVATED: "activated",
|
||||
NM.DeviceState.DEACTIVATING: "deactivating",
|
||||
NM.DeviceState.FAILED: "failed",
|
||||
}.get(self._device.get_state(), "unknown")
|
||||
|
||||
|
||||
class Ethernet(Service):
|
||||
"""A service to manage ethernet devices"""
|
||||
@@ -554,80 +312,125 @@ class Ethernet(Service):
|
||||
def enabled(self) -> bool: ...
|
||||
|
||||
@Property(int, "readable")
|
||||
def speed(self) -> str:
|
||||
speed_mbps = self._device.get_speed()
|
||||
return f"{speed_mbps} Mb/s"
|
||||
|
||||
@Property(str, "readable")
|
||||
def state(self) -> str:
|
||||
return snake_case_to_kebab_case(
|
||||
get_enum_member_name(self._device.get_state(), default="disconnected")
|
||||
)
|
||||
def speed(self) -> int:
|
||||
return self._device.get_speed()
|
||||
|
||||
@Property(str, "readable")
|
||||
def internet(self) -> str:
|
||||
if self._active_connection:
|
||||
return snake_case_to_kebab_case(
|
||||
get_enum_member_name(
|
||||
self._active_connection.get_state(), default="disconnected"
|
||||
)
|
||||
)
|
||||
return "disconnected"
|
||||
|
||||
@Property(str, "readable")
|
||||
def iface(self) -> str:
|
||||
return self._device.get_iface() if self._device else None
|
||||
return {
|
||||
NM.ActiveConnectionState.ACTIVATED: "activated",
|
||||
NM.ActiveConnectionState.ACTIVATING: "activating",
|
||||
NM.ActiveConnectionState.DEACTIVATING: "deactivating",
|
||||
NM.ActiveConnectionState.DEACTIVATED: "deactivated",
|
||||
}.get(
|
||||
self._device.get_active_connection().get_state(),
|
||||
"disconnected",
|
||||
)
|
||||
|
||||
@Property(str, "readable")
|
||||
def icon_name(self) -> str:
|
||||
network = self.internet
|
||||
if network == "activated":
|
||||
return "network-wired-symbolic"
|
||||
|
||||
elif network == "activating":
|
||||
return "network-wired-acquiring-symbolic"
|
||||
|
||||
elif self._device.get_connectivity != NM.ConnectivityState.FULL:
|
||||
return "network-wired-no-route-symbolic"
|
||||
|
||||
return "network-wired-disconnected-symbolic"
|
||||
|
||||
def __init__(self, client: NM.Client, device: NM.DeviceEthernet, **kwargs) -> None:
|
||||
super().__init__(**kwargs)
|
||||
self._client: NM.Client = client
|
||||
self._device: NM.DeviceEthernet = device
|
||||
self._active_connection = None
|
||||
|
||||
self._device.connect(
|
||||
"state-changed", lambda *args: self.on_network_state_changed()
|
||||
for names in (
|
||||
"active-connection",
|
||||
"icon-name",
|
||||
"internet",
|
||||
"speed",
|
||||
"state",
|
||||
):
|
||||
self._device.connect(f"notify::{names}", lambda *_: self.notifier(names))
|
||||
|
||||
self._device.connect("notify::speed", lambda *_: print(_))
|
||||
|
||||
def notifier(self, names):
|
||||
self.notify(names)
|
||||
self.emit("changed")
|
||||
|
||||
|
||||
class NetworkService(Service):
|
||||
"""A service to manage network devices"""
|
||||
|
||||
_instance = None
|
||||
|
||||
def __new__(cls):
|
||||
if cls._instance is None:
|
||||
cls._instance = super().__new__(cls)
|
||||
return cls._instance
|
||||
|
||||
@Signal
|
||||
def device_ready(self) -> None: ...
|
||||
|
||||
def __init__(self, **kwargs):
|
||||
self._client: NM.Client | None = None
|
||||
self.wifi_device: Wifi | None = None
|
||||
self.ethernet_device: Ethernet | None = None
|
||||
super().__init__(**kwargs)
|
||||
NM.Client.new_async(
|
||||
cancellable=None,
|
||||
callback=self._init_network_client,
|
||||
**kwargs,
|
||||
)
|
||||
|
||||
self.update_active_connection()
|
||||
def _init_network_client(self, client: NM.Client, task: Gio.Task, **kwargs):
|
||||
self._client = client
|
||||
wifi_device: NM.DeviceWifi | None = self._get_device(NM.DeviceType.WIFI) # type: ignore
|
||||
ethernet_device: NM.DeviceEthernet | None = self._get_device(
|
||||
NM.DeviceType.ETHERNET
|
||||
)
|
||||
|
||||
def on_network_state_changed(self):
|
||||
"""Called when networking is toggled on/off."""
|
||||
if self.state != "unmanaged":
|
||||
# Re-initialize device when networking is re-enabled
|
||||
self.update_active_connection()
|
||||
else:
|
||||
self._active_connection = None
|
||||
if wifi_device:
|
||||
self.wifi_device = Wifi(self._client, wifi_device)
|
||||
self.emit("device-ready")
|
||||
|
||||
def update_active_connection(self):
|
||||
"""Updates the active connection and connects to its state change signal."""
|
||||
active_connection = self._device.get_active_connection()
|
||||
if active_connection:
|
||||
self._active_connection = active_connection
|
||||
self._active_connection.connect(
|
||||
"state-changed", lambda *args: self.emit("changed")
|
||||
)
|
||||
if ethernet_device:
|
||||
self.ethernet_device = Ethernet(client=self._client, device=ethernet_device)
|
||||
self.emit("device-ready")
|
||||
|
||||
def get_network_stats(self):
|
||||
"""Fetch received and transmitted bytes from the system files"""
|
||||
try:
|
||||
# Read data from /sys/class/net for accurate speed
|
||||
with open(
|
||||
f"/sys/class/net/{self.iface}/statistics/rx_bytes", "r"
|
||||
) as rx_file:
|
||||
rx_bytes = int(rx_file.read().strip())
|
||||
with open(
|
||||
f"/sys/class/net/{self.iface}/statistics/tx_bytes", "r"
|
||||
) as tx_file:
|
||||
tx_bytes = int(tx_file.read().strip())
|
||||
return rx_bytes, tx_bytes
|
||||
except FileNotFoundError:
|
||||
return None, None
|
||||
self.notify("primary-device")
|
||||
|
||||
def _get_device(self, device_type) -> Any:
|
||||
devices: List[NM.Device] = self._client.get_devices() # type: ignore
|
||||
return next(
|
||||
(
|
||||
x
|
||||
for x in devices
|
||||
if x.get_device_type() == device_type
|
||||
and x.get_active_connection() is not None
|
||||
),
|
||||
None,
|
||||
)
|
||||
|
||||
def _get_primary_device(self) -> Literal["wifi", "wired"] | None:
|
||||
if not self._client:
|
||||
return None
|
||||
|
||||
if self._client.get_primary_connection() is None:
|
||||
return "wifi"
|
||||
return (
|
||||
"wifi"
|
||||
if "wireless"
|
||||
in str(self._client.get_primary_connection().get_connection_type())
|
||||
else "wired"
|
||||
if "ethernet"
|
||||
in str(self._client.get_primary_connection().get_connection_type())
|
||||
else None
|
||||
)
|
||||
|
||||
@Property(str, "readable")
|
||||
def primary_device(self) -> Literal["wifi", "wired"] | None:
|
||||
return self._get_primary_device()
|
||||
Reference in new issue
Block a user