feat(astal-menu): NetworkManager control panel; single-ring settings/close buttons
Network expanded view is rewritten into four tabs over a new nmcli backend (backend/nm.py, JSON in/out): - Adapters: each device foldout with editable IPv4 and IPv6 — Automatic/DHCP toggle, address, a linked subnet-mask <-> CIDR pair (edit either, the other follows), gateway; applied via `nmcli connection modify/up`. Manual fields dim when DHCP is on. - Routes: the kernel routing table (read-only; editing kernel routes needs root). - VLAN: pick a parent device + ID to create/bring up a VLAN, list + activate/delete. - DNS: effective servers per device + a per-connection override (servers + ignore-automatic). Verified reads and the write path end-to-end (VLAN add/delete via nmcli; each tab rendered in a harness). Also: flatten the settings-cog MenuButton's inner `button` node so it shows one ring instead of two, and drop the border on the close (x) button. Weather keeps its wttr.in 3-day forecast (unchanged, per request). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XUWCXM4KhjRkwheaA3X7bPfeat/astal-menu
parent
4e5ee4f9ba
commit
d468d66e5f
|
|
@ -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 <con> ipv4/ipv6 method/addresses/gateway/dns
|
||||
set-ip <con> ipv4|ipv6 <method> [addr_cidr] [gateway] [dns_space_sep]
|
||||
routes kernel routing table (read-only)
|
||||
vlans configured VLAN connections
|
||||
vlan-add <parent_dev> <id> [name] create + bring up a VLAN
|
||||
vlan-del <name> delete a VLAN connection
|
||||
vlan-up <name> activate a VLAN connection
|
||||
dns effective DNS servers per device
|
||||
set-dns <con> ipv4|ipv6 <servers_space_sep> <yes|no ignore-auto>
|
||||
"""
|
||||
|
||||
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:]))
|
||||
|
|
@ -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)],
|
||||
)
|
||||
|
|
|
|||
|
|
@ -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; }
|
||||
|
|
|
|||
Loading…
Reference in New Issue