diff --git a/desktopenvs/hyprlua/astal-menu/backend/nm.py b/desktopenvs/hyprlua/astal-menu/backend/nm.py new file mode 100755 index 0000000..337ec13 --- /dev/null +++ b/desktopenvs/hyprlua/astal-menu/backend/nm.py @@ -0,0 +1,212 @@ +#!/usr/bin/env python3 +"""NetworkManager backend for the Network quad's expanded settings. + +Every subcommand prints JSON to stdout ({"ok": bool, ...} for mutations) and +diagnostics to stderr. Reads use `nmcli -g` (raw, unescaped single fields) and +`ip -j`; writes go through `nmcli connection modify/up`, so persistence and polkit +auth are handled by NetworkManager. Kept as one script so the UI has a single, +stable contract (see modules/network.py). + + adapters list manageable devices + their connection + ipconfig ipv4/ipv6 method/addresses/gateway/dns + set-ip ipv4|ipv6 [addr_cidr] [gateway] [dns_space_sep] + routes kernel routing table (read-only) + vlans configured VLAN connections + vlan-add [name] create + bring up a VLAN + vlan-del delete a VLAN connection + vlan-up activate a VLAN connection + dns effective DNS servers per device + set-dns ipv4|ipv6 +""" + +from __future__ import annotations + +import json +import subprocess +import sys + + +def _run(argv: list[str], timeout: int = 20) -> tuple[bool, str, str]: + try: + p = subprocess.run(argv, capture_output=True, text=True, timeout=timeout) + except (FileNotFoundError, subprocess.SubprocessError) as exc: + return False, "", str(exc) + return p.returncode == 0, p.stdout, p.stderr + + +def _get(field: str, con: str) -> str: + ok, out, _ = _run(["nmcli", "-g", field, "connection", "show", con]) + return out.strip() if ok else "" + + +def _list(field: str, con: str) -> list[str]: + raw = _get(field, con) + return [v.strip() for v in raw.replace(",", " ").split() if v.strip()] + + +def _emit(obj) -> int: + print(json.dumps(obj)) + return 0 + + +def _ok(ok: bool, err: str = "") -> int: + print(json.dumps({"ok": bool(ok), "error": err.strip()})) + return 0 if ok else 1 + + +# -- reads ----------------------------------------------------------------- +def adapters() -> int: + ok, out, _ = _run(["nmcli", "-t", "-f", "DEVICE,TYPE,STATE,CONNECTION", "device"]) + rows = [] + if ok: + for line in out.splitlines(): + # DEVICE may contain escaped colons (bt MACs); split from the right on + # the three trailing fields we know are colon-free-ish. + parts = line.split(":") + if len(parts) < 4: + continue + connection = parts[-1] + state = parts[-2] + typ = parts[-3] + device = ":".join(parts[:-3]).replace("\\:", ":") + if typ not in ("ethernet", "wifi"): + continue + rows.append({"device": device, "type": typ, "state": state, + "connection": connection.replace("\\:", ":")}) + return _emit(rows) + + +def ipconfig(con: str) -> int: + def family(fam: str) -> dict: + return { + "method": _get(f"{fam}.method", con) or "auto", + "addresses": _list(f"{fam}.addresses", con), + "gateway": _get(f"{fam}.gateway", con), + "dns": _list(f"{fam}.dns", con), + "ignore_auto_dns": _get(f"{fam}.ignore-auto-dns", con) == "yes", + } + return _emit({"connection": con, "ipv4": family("ipv4"), "ipv6": family("ipv6")}) + + +def routes() -> int: + ok, out, err = _run(["ip", "-j", "route", "show"]) + if not ok: + return _emit([]) + try: + return _emit(json.loads(out)) + except json.JSONDecodeError: + return _emit([]) + + +def vlans() -> int: + ok, out, _ = _run(["nmcli", "-t", "-f", "NAME,TYPE,DEVICE,ACTIVE", "connection", "show"]) + rows = [] + if ok: + for line in out.splitlines(): + parts = line.rsplit(":", 3) + if len(parts) < 4 or parts[1] != "vlan": + continue + name, _typ, device, active = parts + rows.append({"name": name, "device": device, "active": active == "yes"}) + return _emit(rows) + + +def dns() -> int: + """Effective DNS servers, per device, from `nmcli device show`.""" + ok, out, _ = _run(["nmcli", "-t", "-f", + "GENERAL.DEVICE,IP4.DNS,IP6.DNS", "device", "show"]) + devices: dict[str, dict] = {} + cur = None + if ok: + for line in out.splitlines(): + if ":" not in line: + continue + key, _, val = line.partition(":") + val = val.strip() + if key == "GENERAL.DEVICE": + cur = val + devices.setdefault(cur, {"device": cur, "servers": []}) + elif cur and key.startswith(("IP4.DNS", "IP6.DNS")) and val: + devices[cur]["servers"].append(val) + return _emit([d for d in devices.values() if d["servers"]]) + + +# -- writes ---------------------------------------------------------------- +def set_ip(con: str, fam: str, method: str, addr: str = "", + gateway: str = "", dns: str = "") -> int: + if fam not in ("ipv4", "ipv6"): + return _ok(False, "family must be ipv4 or ipv6") + args = ["nmcli", "connection", "modify", con, f"{fam}.method", method] + if method == "manual": + args += [f"{fam}.addresses", addr or ""] + args += [f"{fam}.gateway", gateway or ""] + if dns: + args += [f"{fam}.dns", dns] + else: # auto: clear any manual leftovers so DHCP is clean + args += [f"{fam}.addresses", "", f"{fam}.gateway", ""] + ok, _, err = _run(args) + if ok: + ok, _, err = _run(["nmcli", "connection", "up", con]) + return _ok(ok, err) + + +def set_dns(con: str, fam: str, servers: str, ignore_auto: str) -> int: + if fam not in ("ipv4", "ipv6"): + return _ok(False, "family must be ipv4 or ipv6") + ok, _, err = _run(["nmcli", "connection", "modify", con, + f"{fam}.dns", servers, + f"{fam}.ignore-auto-dns", "yes" if ignore_auto == "yes" else "no"]) + if ok: + ok, _, err = _run(["nmcli", "connection", "up", con]) + return _ok(ok, err) + + +def vlan_add(parent: str, vid: str, name: str = "") -> int: + name = name or f"vlan{vid}" + ok, _, err = _run(["nmcli", "connection", "add", "type", "vlan", + "con-name", name, "dev", parent, "id", str(vid)]) + if ok: + ok, _, err = _run(["nmcli", "connection", "up", name]) + return _ok(ok, err) + + +def vlan_del(name: str) -> int: + ok, _, err = _run(["nmcli", "connection", "delete", name]) + return _ok(ok, err) + + +def vlan_up(name: str) -> int: + ok, _, err = _run(["nmcli", "connection", "up", name]) + return _ok(ok, err) + + +def main(argv: list[str]) -> int: + if not argv: + print("no subcommand", file=sys.stderr) + return 2 + cmd, args = argv[0], argv[1:] + table = { + "adapters": lambda: adapters(), + "ipconfig": lambda: ipconfig(args[0]), + "set-ip": lambda: set_ip(*args), + "routes": lambda: routes(), + "vlans": lambda: vlans(), + "vlan-add": lambda: vlan_add(*args), + "vlan-del": lambda: vlan_del(args[0]), + "vlan-up": lambda: vlan_up(args[0]), + "dns": lambda: dns(), + "set-dns": lambda: set_dns(*args), + } + fn = table.get(cmd) + if fn is None: + print(f"unknown subcommand: {cmd}", file=sys.stderr) + return 2 + try: + return fn() + except TypeError as exc: + print(f"bad arguments for {cmd}: {exc}", file=sys.stderr) + return 2 + + +if __name__ == "__main__": + raise SystemExit(main(sys.argv[1:])) diff --git a/desktopenvs/hyprlua/astal-menu/modules/network.py b/desktopenvs/hyprlua/astal-menu/modules/network.py index 609ce6a..f6c8a99 100644 --- a/desktopenvs/hyprlua/astal-menu/modules/network.py +++ b/desktopenvs/hyprlua/astal-menu/modules/network.py @@ -1,20 +1,36 @@ -"""Network quad: Wi-Fi (via AstalNetwork) plus ip/routes/ports/public-IP and the -manual/DHCP switch (via backend/network.sh). Every section is an independently -toggle-able feature. +"""Network quad. + +Compact: Wi-Fi enable + primary connection/IP (AstalNetwork + backend/network.sh). +Expanded: a small NetworkManager control panel over backend/nm.py, as four tabs — + Adapters : each device foldout with editable IPv4/IPv6 (DHCP toggle, address, + subnet-mask <-> CIDR, gateway); applied via `nmcli connection modify/up` + Routes : the kernel routing table (read-only; editing kernel routes needs root) + VLAN : list / create / activate / delete VLAN connections + DNS : effective servers per device + a per-connection override table +Every tab is an independently toggle-able feature. """ from __future__ import annotations +import ipaddress +import sys + import gi gi.require_version("Gtk", "4.0") from gi.repository import Gtk # noqa: E402 -from lib.proc import run_json, run_text +from lib.proc import run_json from module_base import Feature, ModuleContext, ModuleInstance, ModuleSpec from paths import BACKEND_DIR _NET = str(BACKEND_DIR / "network.sh") +_NM = str(BACKEND_DIR / "nm.py") + + +def _nm(args: list[str], cb) -> None: + """Run a backend/nm.py subcommand expecting JSON on stdout.""" + run_json([sys.executable, _NM, *args], cb) def _primary_ip(data) -> str: @@ -29,6 +45,18 @@ def _primary_ip(data) -> str: return "—" +# -- subnet mask <-> CIDR prefix ------------------------------------------ +def _mask_to_prefix(mask: str) -> int | None: + try: + return ipaddress.IPv4Network(f"0.0.0.0/{mask.strip()}").prefixlen + except (ipaddress.NetmaskValueError, ValueError): + return None + + +def _prefix_to_mask(prefix: int) -> str: + return str(ipaddress.IPv4Network(f"0.0.0.0/{prefix}").netmask) + + class _Compact(Gtk.Box): def __init__(self, ctx: ModuleContext): super().__init__(orientation=Gtk.Orientation.VERTICAL, spacing=8) @@ -74,231 +102,413 @@ class _Compact(Gtk.Box): run_json([_NET, "ip"], lambda ok, d: self._ip.set_text(_primary_ip(d) if ok else "—")) +# -- shared little widgets ------------------------------------------------- +def _clear(box: Gtk.Box) -> None: + child = box.get_first_child() + while child: + box.remove(child) + child = box.get_first_child() + + +def _field(placeholder: str, text: str = "") -> Gtk.Entry: + e = Gtk.Entry(placeholder_text=placeholder, text=text, hexpand=True) + e.add_css_class("net-entry") + return e + + +def _labeled(label: str, widget: Gtk.Widget) -> Gtk.Widget: + row = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8) + lbl = Gtk.Label(label=label, xalign=0.0) + lbl.set_size_request(90, -1) + row.append(lbl) + row.append(widget) + return row + + +def _pill(label: str, cb) -> Gtk.Button: + b = Gtk.Button(label=label) + b.add_css_class("quad-action") + b.connect("clicked", lambda *_: cb()) + return b + + +# -- IPv4 / IPv6 editor for one connection -------------------------------- +class _IPSection(Gtk.Box): + """One address family's editor. IPv4 exposes a linked subnet-mask/CIDR pair; + IPv6 exposes a prefix-length field. Fields are only sensitive in Manual mode.""" + + def __init__(self, con: str, family: str, fam_data: dict, on_result): + super().__init__(orientation=Gtk.Orientation.VERTICAL, spacing=6) + self.con = con + self.family = family + self._on_result = on_result + self.add_css_class("net-view") + + title = "IPv4" if family == "ipv4" else "IPv6" + head = Gtk.CenterBox() + head.set_start_widget(Gtk.Label(label=title, xalign=0.0)) + auto = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8) + auto.append(Gtk.Label(label="Automatic")) + self._dhcp = Gtk.Switch(active=(fam_data.get("method", "auto") != "manual")) + self._dhcp.connect("state-set", self._on_mode) + auto.append(self._dhcp) + head.set_end_widget(auto) + self.append(head) + + addr, prefix = self._split(fam_data.get("addresses", [])) + self._addr = _field("Address", addr) + self._gw = _field("Gateway", fam_data.get("gateway", "")) + + self._fields = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=6) + self._fields.append(_labeled("Address", self._addr)) + if family == "ipv4": + self._guard = False + self._mask = _field("255.255.255.0", _prefix_to_mask(prefix) if prefix else "") + self._cidr = _field("/24", f"/{prefix}" if prefix else "") + self._cidr.set_size_request(70, -1) + self._mask.connect("changed", self._sync_from_mask) + self._cidr.connect("changed", self._sync_from_cidr) + mask_row = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8) + mask_row.append(self._mask) + mask_row.append(self._cidr) + self._fields.append(_labeled("Netmask", mask_row)) + else: + self._prefix = _field("64", str(prefix) if prefix else "") + self._fields.append(_labeled("Prefix", self._prefix)) + self._fields.append(_labeled("Gateway", self._gw)) + apply = Gtk.Button(label=f"Apply {title}") + apply.add_css_class("enable-btn") + apply.connect("clicked", lambda *_: self._apply()) + self._fields.append(apply) + self.append(self._fields) + + self._set_manual_enabled(not self._dhcp.get_active()) + + def _set_manual_enabled(self, on: bool) -> None: + # The reset GTK theme doesn't dim insensitive widgets, so fade the manual + # fields explicitly when DHCP/Automatic is active. + self._fields.set_sensitive(on) + self._fields.set_opacity(1.0 if on else 0.45) + + @staticmethod + def _split(addresses: list[str]) -> tuple[str, int | None]: + if not addresses: + return "", None + addr, _, prefix = addresses[0].partition("/") + try: + return addr, int(prefix) + except ValueError: + return addr, None + + def _on_mode(self, _sw, auto: bool) -> bool: + self._set_manual_enabled(not auto) + return False + + # keep the dotted mask and /CIDR fields in lock-step (IPv4 only) + def _sync_from_mask(self, _e) -> None: + if self._guard: + return + p = _mask_to_prefix(self._mask.get_text()) + if p is not None: + self._guard = True + self._cidr.set_text(f"/{p}") + self._guard = False + + def _sync_from_cidr(self, _e) -> None: + if self._guard: + return + raw = self._cidr.get_text().lstrip("/").strip() + if raw.isdigit() and 0 <= int(raw) <= 32: + self._guard = True + self._mask.set_text(_prefix_to_mask(int(raw))) + self._guard = False + + def _prefixlen(self) -> str: + if self.family == "ipv4": + raw = self._cidr.get_text().lstrip("/").strip() + if raw.isdigit(): + return raw + p = _mask_to_prefix(self._mask.get_text()) + return str(p) if p is not None else "" + return self._prefix.get_text().strip() + + def _apply(self) -> None: + if self._dhcp.get_active(): + args = ["set-ip", self.con, self.family, "auto"] + else: + addr, plen = self._addr.get_text().strip(), self._prefixlen() + cidr = f"{addr}/{plen}" if addr and plen else addr + args = ["set-ip", self.con, self.family, "manual", + cidr, self._gw.get_text().strip(), ""] + _nm(args, lambda ok, d: self._on_result( + ok and isinstance(d, dict) and d.get("ok"), + (d.get("error") if isinstance(d, dict) else "") or "")) + + +class _AdaptersPage(Gtk.Box): + def __init__(self): + super().__init__(orientation=Gtk.Orientation.VERTICAL, spacing=6) + self._status = Gtk.Label(xalign=0.0) + self._status.add_css_class("net-ip") + self._status.set_visible(False) + self.append(self._status) + self._list = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=8) + scroller = Gtk.ScrolledWindow(vexpand=True, hscrollbar_policy=Gtk.PolicyType.NEVER) + scroller.set_child(self._list) + self.append(scroller) + self._reload() + + def _reload(self) -> None: + _nm(["adapters"], self._populate) + + def _populate(self, ok: bool, adapters) -> None: + _clear(self._list) + if not ok or not isinstance(adapters, list) or not adapters: + self._list.append(Gtk.Label(label="No manageable adapters", xalign=0.0)) + return + for a in adapters: + self._list.append(self._adapter_row(a)) + + def _adapter_row(self, a: dict) -> Gtk.Widget: + icon = "" if a.get("type") == "wifi" else "" # nf wifi / ethernet + exp = Gtk.Expander( + label=f"{icon} {a['device']} · {a.get('connection') or '—'} [{a.get('state')}]") + exp.add_css_class("net-adapter") + body = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=10) + body.set_margin_top(8) + con = a.get("connection") + if not con: + body.append(Gtk.Label(label="No connection profile", xalign=0.0)) + else: + loading = Gtk.Label(label="Loading…", xalign=0.0) + body.append(loading) + _nm(["ipconfig", con], lambda ok, d, b=body, l=loading, c=con: + self._fill(ok, d, b, l, c)) + exp.set_child(body) + return exp + + def _fill(self, ok, data, body, loading, con) -> None: + body.remove(loading) + if not ok or not isinstance(data, dict): + body.append(Gtk.Label(label="Could not read config", xalign=0.0)) + return + result = Gtk.Label(xalign=0.0) + result.set_visible(False) + + def on_result(good: bool, err: str) -> None: + result.set_text("Applied" if good else f"Failed: {err or 'error'}") + result.set_visible(True) + + body.append(_IPSection(con, "ipv4", data.get("ipv4", {}), on_result)) + body.append(Gtk.Separator()) + body.append(_IPSection(con, "ipv6", data.get("ipv6", {}), on_result)) + body.append(result) + + +class _RoutesPage(Gtk.Box): + def __init__(self): + super().__init__(orientation=Gtk.Orientation.VERTICAL, spacing=4) + header = Gtk.CenterBox() + header.set_start_widget(Gtk.Label(label="Kernel routing table", xalign=0.0)) + header.set_end_widget(_pill(" Refresh", self._reload)) + self.append(header) + self._list = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=2) + scroller = Gtk.ScrolledWindow(vexpand=True, hscrollbar_policy=Gtk.PolicyType.NEVER) + scroller.set_child(self._list) + self.append(scroller) + self._reload() + + def _reload(self) -> None: + _nm(["routes"], self._populate) + + def _populate(self, ok: bool, routes) -> None: + _clear(self._list) + if not ok or not isinstance(routes, list) or not routes: + self._list.append(Gtk.Label(label="No routes", xalign=0.0)) + return + for r in routes: + self._list.append(Gtk.Label(label=self._fmt(r), xalign=0.0, + selectable=True, wrap=True)) + + @staticmethod + def _fmt(r: dict) -> str: + dst = r.get("dst", "?") + via = f" via {r['gateway']}" if r.get("gateway") else "" + dev = f" dev {r['dev']}" if r.get("dev") else "" + metric = f" metric {r['metric']}" if r.get("metric") is not None else "" + proto = f" ({r['proto']})" if r.get("proto") else "" + return f"{dst}{via}{dev}{metric}{proto}" + + +class _VlanPage(Gtk.Box): + def __init__(self): + super().__init__(orientation=Gtk.Orientation.VERTICAL, spacing=8) + self._adapters: list[dict] = [] + + add = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=6) + self._parent = Gtk.DropDown.new_from_strings(["…"]) + self._vid = _field("VLAN ID") + self._vid.set_size_request(90, -1) + add.append(self._parent) + add.append(self._vid) + add.append(_pill("Add", self._add)) + self.append(_labeled("New VLAN", add)) + + self._status = Gtk.Label(xalign=0.0) + self._status.add_css_class("net-ip") + self._status.set_visible(False) + self.append(self._status) + + self._list = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=4) + scroller = Gtk.ScrolledWindow(vexpand=True, hscrollbar_policy=Gtk.PolicyType.NEVER) + scroller.set_child(self._list) + self.append(scroller) + + _nm(["adapters"], self._set_adapters) + self._reload() + + def _set_adapters(self, ok, adapters) -> None: + self._adapters = adapters if ok and isinstance(adapters, list) else [] + names = [a["device"] for a in self._adapters] or ["(no devices)"] + self._parent.set_model(Gtk.StringList.new(names)) + + def _reload(self) -> None: + _nm(["vlans"], self._populate) + + def _populate(self, ok, vlans) -> None: + _clear(self._list) + if not ok or not isinstance(vlans, list) or not vlans: + self._list.append(Gtk.Label(label="No VLANs configured", xalign=0.0)) + return + for v in vlans: + self._list.append(self._vlan_row(v)) + + def _vlan_row(self, v: dict) -> Gtk.Widget: + row = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8) + row.add_css_class("net-row") + state = "active" if v.get("active") else "inactive" + row.append(Gtk.Label(label=f"{v['name']} · {v.get('device') or '—'} [{state}]", + xalign=0.0, hexpand=True)) + if not v.get("active"): + row.append(_pill("Up", lambda n=v["name"]: self._act("vlan-up", n))) + row.append(_pill("Delete", lambda n=v["name"]: self._act("vlan-del", n))) + return row + + def _add(self) -> None: + idx = self._parent.get_selected() + vid = self._vid.get_text().strip() + if idx < 0 or idx >= len(self._adapters) or not vid.isdigit(): + self._flash("Pick a parent device and numeric VLAN ID") + return + parent = self._adapters[idx]["device"] + _nm(["vlan-add", parent, vid], self._after_write) + + def _act(self, cmd: str, name: str) -> None: + _nm([cmd, name], self._after_write) + + def _after_write(self, ok, d) -> None: + good = ok and isinstance(d, dict) and d.get("ok") + self._flash("Done" if good else + f"Failed: {(d.get('error') if isinstance(d, dict) else '') or 'error'}") + self._reload() + + def _flash(self, msg: str) -> None: + self._status.set_text(msg) + self._status.set_visible(True) + + +class _DnsPage(Gtk.Box): + def __init__(self): + super().__init__(orientation=Gtk.Orientation.VERTICAL, spacing=8) + self._adapters: list[dict] = [] + + self.append(Gtk.Label(label="Effective servers", xalign=0.0)) + self._effective = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=2) + self.append(self._effective) + + self.append(Gtk.Separator()) + self.append(Gtk.Label(label="Override", xalign=0.0)) + self._conn = Gtk.DropDown.new_from_strings(["…"]) + self.append(_labeled("Connection", self._conn)) + self._servers = _field("1.1.1.1 8.8.8.8") + self.append(_labeled("Servers", self._servers)) + ignore = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8) + ignore.append(Gtk.Label(label="Ignore automatic", hexpand=True, xalign=0.0)) + self._ignore = Gtk.Switch(active=True) + ignore.append(self._ignore) + self.append(ignore) + self.append(_pill("Apply DNS", self._apply)) + self._status = Gtk.Label(xalign=0.0) + self._status.add_css_class("net-ip") + self._status.set_visible(False) + self.append(self._status) + + _nm(["adapters"], self._set_adapters) + _nm(["dns"], self._show_effective) + + def _set_adapters(self, ok, adapters) -> None: + self._adapters = [a for a in (adapters or []) if a.get("connection")] \ + if ok and isinstance(adapters, list) else [] + names = [f"{a['connection']} ({a['device']})" for a in self._adapters] or ["(none)"] + self._conn.set_model(Gtk.StringList.new(names)) + + def _show_effective(self, ok, data) -> None: + _clear(self._effective) + if not ok or not isinstance(data, list) or not data: + self._effective.append(Gtk.Label(label="—", xalign=0.0)) + return + for d in data: + self._effective.append(Gtk.Label( + label=f"{d['device']}: {', '.join(d.get('servers', []))}", + xalign=0.0, selectable=True)) + + def _apply(self) -> None: + idx = self._conn.get_selected() + if idx < 0 or idx >= len(self._adapters): + self._flash("Pick a connection") + return + con = self._adapters[idx]["connection"] + servers = self._servers.get_text().strip() + ign = "yes" if self._ignore.get_active() else "no" + _nm(["set-dns", con, "ipv4", servers, ign], self._after) + + def _after(self, ok, d) -> None: + good = ok and isinstance(d, dict) and d.get("ok") + self._flash("Applied" if good else + f"Failed: {(d.get('error') if isinstance(d, dict) else '') or 'error'}") + _nm(["dns"], self._show_effective) + + def _flash(self, msg: str) -> None: + self._status.set_text(msg) + self._status.set_visible(True) + + class _Expanded(Gtk.Box): def __init__(self, ctx: ModuleContext): super().__init__(orientation=Gtk.Orientation.VERTICAL, spacing=8) self.add_css_class("net-view") self.ctx = ctx - self.net = ctx.services.network self.stack = Gtk.Stack(vexpand=True) switcher = Gtk.StackSwitcher(stack=self.stack) switcher.add_css_class("net-switcher") self.append(switcher) self.append(self.stack) - self._build_pages() - def _build_pages(self) -> None: - child = self.stack.get_first_child() - while child: - self.stack.remove(child) - child = self.stack.get_first_child() - f = self.ctx.feature - if f("wifi", True): - self.stack.add_titled(self._wifi_page(), "wifi", "Wi-Fi") - if f("ip_config", True): - self.stack.add_titled(self._ip_page(), "ip", "IP") + f = ctx.feature + if f("adapters", True): + self.stack.add_titled(_AdaptersPage(), "adapters", "Adapters") if f("routes", True): - self.stack.add_titled(self._list_page("routes", self._fmt_route), "routes", "Routes") - if f("ports", True): - self.stack.add_titled(self._list_page("ports", self._fmt_port), "ports", "Ports") - if f("public_ip", True): - self.stack.add_titled(self._pubip_page(), "pub", "Public IP") - - # -- Wi-Fi ------------------------------------------------------------- - def _wifi_page(self) -> Gtk.Widget: - box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=6) - header = Gtk.CenterBox() - header.set_start_widget(Gtk.Label(label="Networks", xalign=0.0)) - scan = Gtk.Button(label=" Scan") - scan.add_css_class("quad-action") - scan.connect("clicked", lambda *_: self._scan()) - header.set_end_widget(scan) - box.append(header) - - self._ap_list = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=4) - scroller = Gtk.ScrolledWindow(vexpand=True, hscrollbar_policy=Gtk.PolicyType.NEVER) - scroller.set_child(self._ap_list) - box.append(scroller) - - wifi = self.net.get_wifi() if self.net else None - if wifi: - wifi.connect("notify::access-points", lambda *_: self._refresh_aps()) - wifi.connect("notify::active-access-point", lambda *_: self._refresh_aps()) - self._refresh_aps() - return box - - def _scan(self) -> None: - wifi = self.net.get_wifi() if self.net else None - if wifi: - wifi.scan() - - def _refresh_aps(self) -> None: - child = self._ap_list.get_first_child() - while child: - self._ap_list.remove(child) - child = self._ap_list.get_first_child() - wifi = self.net.get_wifi() if self.net else None - if not wifi: - self._ap_list.append(Gtk.Label(label="No Wi-Fi device")) - return - active = wifi.get_active_access_point() - seen = {} - for ap in wifi.get_access_points(): - ssid = ap.get_ssid() - if not ssid: - continue - if ssid not in seen or ap.get_strength() > seen[ssid].get_strength(): - seen[ssid] = ap - for ap in sorted(seen.values(), key=lambda a: -a.get_strength()): - is_active = active is not None and ap.get_ssid() == active.get_ssid() - self._ap_list.append(self._ap_row(ap, is_active)) - - def _ap_row(self, ap, is_active: bool) -> Gtk.Widget: - outer = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=2) - row = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=10) - row.add_css_class("net-row") - row.append(Gtk.Image.new_from_icon_name(ap.get_icon_name() or "network-wireless-symbolic")) - lbl = Gtk.Label(label=ap.get_ssid(), xalign=0.0, hexpand=True) - row.append(lbl) - row.append(Gtk.Label(label=f"{ap.get_strength()}%")) - if ap.get_requires_password(): - row.append(Gtk.Label(label="")) # lock glyph - - btn = Gtk.Button(label="Disconnect" if is_active else "Connect") - btn.add_css_class("quad-action") - row.append(btn) - outer.append(row) - - if is_active: - btn.connect("clicked", lambda *_: run_text([_NET, "wifi-disconnect"], self._ignore)) - elif ap.get_requires_password(): - reveal = Gtk.Revealer(transition_type=Gtk.RevealerTransitionType.SLIDE_DOWN) - pw_row = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=6) - entry = Gtk.PasswordEntry(show_peek_icon=True, hexpand=True) - go = Gtk.Button(label="Join") - go.add_css_class("quad-action") - pw_row.append(entry) - pw_row.append(go) - reveal.set_child(pw_row) - outer.append(reveal) - btn.connect("clicked", lambda *_: reveal.set_reveal_child(not reveal.get_reveal_child())) - go.connect("clicked", lambda *_, s=ap.get_ssid(): - run_text([_NET, "wifi-connect", s, entry.get_text()], self._ignore)) - else: - btn.connect("clicked", lambda *_, s=ap.get_ssid(): - run_text([_NET, "wifi-connect", s], self._ignore)) - return outer - - # -- IP config --------------------------------------------------------- - def _ip_page(self) -> Gtk.Widget: - box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=8) - self._ip_current = Gtk.Label(label="…", xalign=0.0) - self._ip_current.add_css_class("net-ip") - box.append(self._ip_current) - - mode = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8) - mode.append(Gtk.Label(label="Manual")) - self._manual = Gtk.Switch() - self._manual.connect("state-set", self._on_mode) - mode.append(self._manual) - box.append(mode) - - self._manual_fields = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=6) - self._addr = self._field("Address/CIDR (e.g. 192.168.1.50/24)") - self._gw = self._field("Gateway") - self._dns = self._field("DNS") - for w in (self._addr, self._gw, self._dns): - self._manual_fields.append(w) - apply = Gtk.Button(label="Apply") - apply.add_css_class("enable-btn") - apply.connect("clicked", lambda *_: self._apply_manual()) - self._manual_fields.append(apply) - self._manual_fields.set_sensitive(False) - box.append(self._manual_fields) - - run_json([_NET, "ip"], lambda ok, d: - self._ip_current.set_text(_primary_ip(d) if ok else "—")) - return box - - @staticmethod - def _field(placeholder: str) -> Gtk.Entry: - e = Gtk.Entry(placeholder_text=placeholder) - e.add_css_class("net-entry") - return e - - def _on_mode(self, _sw, manual: bool) -> bool: - self._manual_fields.set_sensitive(manual) - if not manual: - run_text([_NET, "set-dhcp"], self._ignore) - return False - - def _apply_manual(self) -> None: - run_text([_NET, "set-manual", self._addr.get_text(), - self._gw.get_text(), self._dns.get_text()], self._ignore) - - # -- generic list pages ------------------------------------------------ - def _list_page(self, sub: str, fmt) -> Gtk.Widget: - box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=4) - listing = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=2) - scroller = Gtk.ScrolledWindow(vexpand=True, hscrollbar_policy=Gtk.PolicyType.NEVER) - scroller.set_child(listing) - refresh = Gtk.Button(label=" Refresh") - refresh.add_css_class("quad-action") - refresh.connect("clicked", lambda *_: self._load_list(sub, fmt, listing)) - box.append(refresh) - box.append(scroller) - self._load_list(sub, fmt, listing) - return box - - def _load_list(self, sub: str, fmt, listing: Gtk.Box) -> None: - child = listing.get_first_child() - while child: - listing.remove(child) - child = listing.get_first_child() - - def done(ok, data): - if not ok or not isinstance(data, list): - listing.append(Gtk.Label(label="unavailable", xalign=0.0)) - return - for item in data: - listing.append(Gtk.Label(label=fmt(item), xalign=0.0, - selectable=True, wrap=True)) - run_json([_NET, sub], done) - - @staticmethod - def _fmt_route(r: dict) -> str: - dst = r.get("dst", "?") - via = f" via {r['gateway']}" if r.get("gateway") else "" - dev = f" dev {r['dev']}" if r.get("dev") else "" - return f"{dst}{via}{dev}" - - @staticmethod - def _fmt_port(p: dict) -> str: - return f"{p.get('proto', '?'):5} {p.get('local', '')}" - - # -- public IP --------------------------------------------------------- - def _pubip_page(self) -> Gtk.Widget: - box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=8, - valign=Gtk.Align.CENTER) - label = Gtk.Label(label="—", selectable=True) - label.add_css_class("pubip") - btn = Gtk.Button(label="Query public IP") - btn.add_css_class("enable-btn") - btn.connect("clicked", lambda *_: run_text( - [_NET, "pubip"], lambda ok, out, err: label.set_text(out.strip() if ok else "unavailable"))) - box.append(label) - box.append(btn) - return box - - @staticmethod - def _ignore(*_a) -> None: - pass + self.stack.add_titled(_RoutesPage(), "routes", "Routes") + if f("vlan", True): + self.stack.add_titled(_VlanPage(), "vlan", "VLAN") + if f("dns", True): + self.stack.add_titled(_DnsPage(), "dns", "DNS") def build(ctx: ModuleContext) -> ModuleInstance: # Feature toggles rebuild the whole card via QuadCard, so _Expanded is recreated - # with the enabled pages — no in-place page rebuild needed here. + # with the enabled tabs — no in-place page rebuild needed here. compact = _Compact(ctx) expanded = _Expanded(ctx) return ModuleInstance(compact=compact, expanded=expanded) @@ -310,9 +520,8 @@ SPEC = ModuleSpec( icon="", # nf-md-lan build=build, default_enabled=True, - features=[Feature("wifi", "Wi-Fi", True), - Feature("ip_config", "IP / DHCP config", True), + features=[Feature("adapters", "Adapters", True), Feature("routes", "Routes", True), - Feature("ports", "Open ports", True), - Feature("public_ip", "Public IP query", True)], + Feature("vlan", "VLAN", True), + Feature("dns", "DNS", True)], ) diff --git a/desktopenvs/hyprlua/astal-menu/style/style.css b/desktopenvs/hyprlua/astal-menu/style/style.css index 0d17849..63a3c7c 100644 --- a/desktopenvs/hyprlua/astal-menu/style/style.css +++ b/desktopenvs/hyprlua/astal-menu/style/style.css @@ -84,6 +84,17 @@ drawingarea { .quad-action:active, button:checked.quad-action { background: @accent; color: @bg; border-color: @accent; } +/* A MenuButton (the settings cog) wraps an inner `button` node that would render a + * second ring inside the outer .quad-action pill — flatten it so only one border + * shows. */ +.quad-action > button { + border: none; + background: transparent; + padding: 0; + min-width: 0; + min-height: 0; +} + .quad-disabled { color: @text; opacity: 0.7; } .enable-btn { border-color: @accent; } @@ -175,6 +186,8 @@ button:checked.quad-action { background: @accent; color: @bg; border-color: @acc .net-ip { color: @text; opacity: 0.85; font-size: 11pt; } .net-switcher { margin-bottom: 8px; } +.net-adapter { padding: 4px 0; } +.net-adapter > box { padding-left: 6px; } .net-entry { border: 2px solid @violet; border-radius: 12px; @@ -198,7 +211,7 @@ scrollbar slider:hover { background: @accent; } .panel { padding: 2px; } .close-btn { color: @text; background: @bg; - border: 2px solid @accent; border-radius: 20px; + border: none; border-radius: 20px; min-width: 34px; min-height: 34px; margin: 2px 4px; } .close-btn:hover { background: @accent; color: @bg; }