321 lines
12 KiB
Python
321 lines
12 KiB
Python
"""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.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import gi
|
|
|
|
gi.require_version("Gtk", "4.0")
|
|
from gi.repository import Gtk # noqa: E402
|
|
|
|
from lib.proc import run_json, run_text
|
|
from module_base import Feature, ModuleContext, ModuleInstance, ModuleSpec
|
|
from paths import BACKEND_DIR
|
|
|
|
_NET = str(BACKEND_DIR / "network.sh")
|
|
|
|
|
|
def _primary_ip(data) -> str:
|
|
if not isinstance(data, list):
|
|
return "—"
|
|
for iface in data:
|
|
if iface.get("ifname") == "lo":
|
|
continue
|
|
for a in iface.get("addr_info", []):
|
|
if a.get("family") == "inet":
|
|
return f"{a['local']}/{a.get('prefixlen', '')} ({iface.get('ifname')})"
|
|
return "—"
|
|
|
|
|
|
class _Compact(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
|
|
|
|
row = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8)
|
|
self._primary = Gtk.Label(label="…", xalign=0.0, hexpand=True)
|
|
row.append(Gtk.Label(label="")) # nf wifi
|
|
row.append(self._primary)
|
|
wifi = self.net.get_wifi() if self.net else None
|
|
self._wifi_switch = Gtk.Switch(active=bool(wifi and wifi.get_enabled()))
|
|
self._wifi_switch.set_tooltip_text("Wi-Fi")
|
|
self._wifi_switch.connect("state-set", self._on_wifi_toggle)
|
|
row.append(self._wifi_switch)
|
|
self.append(row)
|
|
|
|
self._ip = Gtk.Label(label="", xalign=0.0)
|
|
self._ip.add_css_class("net-ip")
|
|
self.append(self._ip)
|
|
|
|
if self.net:
|
|
self.net.connect("notify::primary", lambda *_: self.refresh())
|
|
if wifi:
|
|
wifi.connect("notify::ssid", lambda *_: self.refresh())
|
|
self.refresh()
|
|
|
|
def _on_wifi_toggle(self, _sw, value: bool) -> bool:
|
|
wifi = self.net.get_wifi() if self.net else None
|
|
if wifi:
|
|
wifi.set_enabled(value)
|
|
return False
|
|
|
|
def refresh(self) -> None:
|
|
wifi = self.net.get_wifi() if self.net else None
|
|
if wifi and wifi.get_active_access_point():
|
|
self._primary.set_text(f"{wifi.get_ssid() or '?'} · {wifi.get_strength()}%")
|
|
elif self.net and self.net.get_wired() and self.net.get_wired().get_internet():
|
|
self._primary.set_text("Wired")
|
|
else:
|
|
self._primary.set_text("Disconnected")
|
|
run_json([_NET, "ip"], lambda ok, d: self._ip.set_text(_primary_ip(d) if ok else "—"))
|
|
|
|
|
|
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")
|
|
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
|
|
|
|
def on_settings_changed(self) -> None:
|
|
self._build_pages()
|
|
|
|
|
|
def build(ctx: ModuleContext) -> ModuleInstance:
|
|
compact = _Compact(ctx)
|
|
expanded = _Expanded(ctx)
|
|
ctx.on_settings_changed(expanded.on_settings_changed)
|
|
return ModuleInstance(compact=compact, expanded=expanded)
|
|
|
|
|
|
SPEC = ModuleSpec(
|
|
id="network",
|
|
title="Network",
|
|
icon="", # nf-md-lan
|
|
build=build,
|
|
default_enabled=True,
|
|
features=[Feature("wifi", "Wi-Fi", True),
|
|
Feature("ip_config", "IP / DHCP config", True),
|
|
Feature("routes", "Routes", True),
|
|
Feature("ports", "Open ports", True),
|
|
Feature("public_ip", "Public IP query", True)],
|
|
)
|