Something went wrong. Try again.
A Material 3 inspired desktop shell for Niri and Hyprland created with Ignis.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111from __future__ import annotations
import loggingimport socketfrom pathlib import Path
from gi.repository import Gio, GLib
from .model import Connectivity, EthernetLinkfrom .notifier import ChangeNotifier
log = logging.getLogger("exo.network")
SYSFS_NET = Path("/sys/class/net")NETLINK_ROUTE = 0RTMGRP_LINK = 1ARPHRD_ETHER = "1"
CONNECTIVITY_BY_GIO = { Gio.NetworkConnectivity.LOCAL: Connectivity.LOCAL_ONLY, Gio.NetworkConnectivity.LIMITED: Connectivity.LIMITED, Gio.NetworkConnectivity.PORTAL: Connectivity.PORTAL, Gio.NetworkConnectivity.FULL: Connectivity.FULL,}
def _read(path: Path) -> str | None: try: return path.read_text().strip() except OSError: return None
def read_ethernet(root: Path = SYSFS_NET) -> tuple[EthernetLink, ...]: links = [] for entry in sorted(root.iterdir()): is_physical_ethernet = ( _read(entry / "type") == ARPHRD_ETHER and not (entry / "wireless").exists() and not (entry / "phy80211").exists() and "/virtual/" not in str(entry.resolve()) ) if not is_physical_ethernet: continue
carrier = _read(entry / "carrier") connected = carrier == "1" if carrier is not None else _read(entry / "operstate") == "up" speed = _read(entry / "speed") links.append( EthernetLink( interface=entry.name, connected=connected, speed_mbps=int(speed) if connected and speed and int(speed) > 0 else None, mac=_read(entry / "address"), ) ) return tuple(links)
class LinkMonitor(ChangeNotifier): def __init__(self, sysfs: Path = SYSFS_NET, watch_netlink: bool = True): super().__init__() self._sysfs = sysfs self._netlink: socket.socket | None = None self.ethernet: tuple[EthernetLink, ...] = read_ethernet(sysfs) self.connectivity = Connectivity.UNKNOWN
if watch_netlink: self._watch_netlink() self._watch_connectivity()
def _watch_netlink(self): try: sock = socket.socket( socket.AF_NETLINK, socket.SOCK_RAW | socket.SOCK_NONBLOCK, NETLINK_ROUTE ) sock.bind((0, RTMGRP_LINK)) except OSError as error: log.warning("netlink unavailable, ethernet state will not update live: %s", error) return self._netlink = sock GLib.io_add_watch(sock.fileno(), GLib.PRIORITY_DEFAULT, GLib.IOCondition.IN, self._on_netlink)
def _on_netlink(self, _fd, _condition): while True: try: self._netlink.recv(65536) except BlockingIOError: break self.refresh_ethernet() return GLib.SOURCE_CONTINUE
def refresh_ethernet(self): current = read_ethernet(self._sysfs) if current != self.ethernet: self.ethernet = current self.notify_changed()
def _watch_connectivity(self): monitor = Gio.NetworkMonitor.get_default() if monitor is None: return monitor.connect("notify::connectivity", lambda *_: self._on_connectivity(monitor)) self._on_connectivity(monitor)
def _on_connectivity(self, monitor: Gio.NetworkMonitor): current = CONNECTIVITY_BY_GIO.get(monitor.get_connectivity(), Connectivity.UNKNOWN) if current != self.connectivity: self.connectivity = current self.notify_changed()