Something went wrong. Try again.
A Material 3 inspired desktop shell for Niri and Hyprland created with Ignis.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142from __future__ import annotations
import loggingimport os
from gi.repository import Gio, GLib
from .backends import Backendfrom .link import LinkMonitorfrom .model import ( Capabilities, ErrorCode, NetworkError, NetworkSnapshot, WifiState,)from .notifier import ChangeNotifier
log = logging.getLogger("exo.network")
NM_BUS_NAME = "org.freedesktop.NetworkManager"
class NetworkController(ChangeNotifier): def __init__(self, backend: Backend | None, links: LinkMonitor): super().__init__() self._backend = backend self._links = links links.connect("changed", lambda *_: self.notify_changed()) if backend is not None: backend.connect("changed", lambda *_: self.notify_changed())
@property def snapshot(self) -> NetworkSnapshot: backend = self._backend return NetworkSnapshot( backend=backend.name if backend else "", wifi=backend.state() if backend else WifiState(), ethernet=self._links.ethernet, connectivity=self._links.connectivity, capabilities=backend.capabilities if backend else Capabilities(), )
def _require_backend(self) -> Backend: if self._backend is None: raise NetworkError(ErrorCode.UNAVAILABLE, "no Wi-Fi backend") return self._backend
async def set_wifi_enabled(self, enabled: bool) -> None: await self._require_backend().set_wifi_enabled(enabled)
async def scan(self) -> None: await self._require_backend().scan()
async def connect_to(self, network_id: str, passphrase: str | None = None) -> None: await self._require_backend().connect_to(network_id, passphrase)
async def disconnect_from(self, network_id: str) -> None: await self._require_backend().disconnect_from(network_id)
async def forget(self, network_id: str) -> None: await self._require_backend().forget(network_id)
async def set_auto_connect(self, network_id: str, enabled: bool) -> None: await self._require_backend().set_auto_connect(network_id, enabled)
def choose_backend(setting: str, networkmanager_running: bool, networkmanager_usable: bool) -> str | None: setting = setting.strip().lower() if setting == "none": return None if setting == "iwd": return "iwd" if setting == "networkmanager": return "networkmanager" if networkmanager_usable else None if networkmanager_running and networkmanager_usable: return "networkmanager" return "iwd"
def _name_has_owner(bus: Gio.DBusConnection, name: str) -> bool: reply = bus.call_sync( "org.freedesktop.DBus", "/org/freedesktop/DBus", "org.freedesktop.DBus", "NameHasOwner", GLib.Variant("(s)", (name,)), None, Gio.DBusCallFlags.NONE, -1, None, ) return reply.unpack()[0]
def _networkmanager_service(): try: from ignis.services.network import NetworkService
return NetworkService.get_default() except Exception as error: # noqa: BLE001 log.info("NetworkManager service unavailable: %s", error) return None
def _build_controller() -> NetworkController | None: setting = os.environ.get("EXO_NETWORK_BACKEND", "auto") try: bus = Gio.bus_get_sync(Gio.BusType.SYSTEM, None) except GLib.Error as error: log.warning("system bus unavailable: %s", error.message) bus = None
service = _networkmanager_service() if setting.strip().lower() != "iwd" else None running = bus is not None and _name_has_owner(bus, NM_BUS_NAME) kind = choose_backend(setting, running, service is not None) if kind is None: return None
backend: Backend | None = None if kind == "networkmanager": from .backends.nm import NetworkManagerBackend
backend = NetworkManagerBackend(service) elif bus is not None: from .backends.iwd import IwdBackend
backend = IwdBackend(bus) return NetworkController(backend, LinkMonitor())
_controller: NetworkController | None = None_resolved = False
def get_network_controller() -> NetworkController | None: global _controller, _resolved if not _resolved: _controller = _build_controller() _resolved = True return _controller