Wrapping the whole page in a ScrolledWindow meant the sliders themselves could scroll out of view, which was the wrong shape entirely. The page is plain stacked cards now, spaced like the home view, and nothing scrolls unless more than four applications are playing - only that list gets a scroller. The 'unplugged' marker was appended to the device name and ellipsis ate it on long names; it is a right-aligned tag now. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Cn3dHfbgNYdbZvzb1b3GQt
974 lines
37 KiB
Python
Executable file
974 lines
37 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""Sakura Line - control panel.
|
|
|
|
A real GTK window: persistent, with live sliders and instant view switching.
|
|
rofi has to relaunch for every menu, which is what made the old panel flicker.
|
|
|
|
Escape steps back a view, or closes the panel from the main view.
|
|
"""
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
|
|
import gi
|
|
gi.require_version("Gtk", "3.0")
|
|
gi.require_version("Gdk", "3.0")
|
|
from gi.repository import Gtk, Gdk, GLib # noqa: E402
|
|
|
|
CSS = b"""
|
|
/* Glyphs come from the Nerd Font. Without an explicit family, GTK uses the
|
|
desktop font (Noto Sans), which lacks them, and Pango's per-glyph fallback
|
|
picks a font that ignores the css colour - that is why icons rendered black. */
|
|
.icon, .tile-icon, .pw, .rowbtn, .volbtn, .title,
|
|
.icon label, .pw label, .rowbtn label, .volbtn label, .tile label {
|
|
font-family: "JetBrainsMono Nerd Font", "Symbols Nerd Font", monospace;
|
|
}
|
|
.pw label, .rowbtn label, .volbtn label { color: #e9eefb; }
|
|
.pw:hover label { color: #e9eefb; }
|
|
|
|
window.panel { background-color: transparent; }
|
|
.frame {
|
|
background-color: rgba(15, 13, 23, 0.88);
|
|
border: 1px solid rgba(146, 185, 234, 0.16);
|
|
border-radius: 18px;
|
|
padding: 20px;
|
|
}
|
|
|
|
.title { font-weight: bold; font-size: 15px; color: #e9eefb; }
|
|
.subtle { color: #8d85a4; font-size: 11px; }
|
|
.section { color: #92b9ea; font-size: 10px; font-weight: bold; letter-spacing: 1px; }
|
|
.hint {
|
|
color: #6f6889;
|
|
font-size: 9px;
|
|
font-weight: bold;
|
|
}
|
|
.tile-icon { font-size: 20px; color: #e9eefb; }
|
|
.tile-name { font-size: 12px; color: #e9eefb; }
|
|
.tile.active .tile-icon { color: #92b9ea; }
|
|
.tile.active .tile-name { color: #e9eefb; }
|
|
.tile.active .subtle { color: #9dbdea; }
|
|
|
|
.card {
|
|
background-color: rgba(34, 28, 51, 0.66);
|
|
border: 1px solid rgba(146, 185, 234, 0.08);
|
|
border-radius: 14px;
|
|
padding: 12px;
|
|
}
|
|
.tile {
|
|
background-image: none;
|
|
background-color: rgba(38, 32, 58, 0.78);
|
|
border: 1px solid rgba(146, 185, 234, 0.10);
|
|
border-radius: 14px;
|
|
color: #e9eefb;
|
|
padding: 14px 6px;
|
|
}
|
|
.tile:hover { background-color: rgba(58, 51, 80, 0.9); }
|
|
.tile.active {
|
|
background-color: rgba(146, 185, 234, 0.14);
|
|
border-color: rgba(146, 185, 234, 0.55);
|
|
}
|
|
.rowbtn {
|
|
background-image: none;
|
|
background-color: transparent;
|
|
border: none;
|
|
border-radius: 10px;
|
|
color: #e9eefb;
|
|
font-size: 14px;
|
|
padding: 10px 12px;
|
|
}
|
|
/* only the speaker/mic buttons beside a slider, not every list row */
|
|
.volbtn {
|
|
background-image: none;
|
|
background-color: transparent;
|
|
border: none;
|
|
border-radius: 10px;
|
|
color: #e9eefb;
|
|
font-size: 19px;
|
|
padding: 6px 10px;
|
|
}
|
|
.volbtn:hover { background-color: rgba(146, 185, 234, 0.17); }
|
|
.rowbtn:hover { background-color: rgba(146, 185, 234, 0.17); }
|
|
.rowbtn.chosen { background-color: rgba(146, 185, 234, 0.22); color: #cfe0f7; }
|
|
.rowbtn.unavail { color: #6f6889; }
|
|
.rowbtn.unavail label { color: #6f6889; }
|
|
.pw {
|
|
background-image: none;
|
|
background-color: rgba(38, 32, 58, 0.78);
|
|
border: 1px solid rgba(146, 185, 234, 0.10);
|
|
border-radius: 12px;
|
|
color: #e9eefb;
|
|
font-size: 20px;
|
|
padding: 13px 8px;
|
|
}
|
|
.pw:hover { background-color: rgba(146, 185, 234, 0.22); }
|
|
.pw.danger:hover { background-color: rgba(207, 106, 128, 0.32); color: #ffe9ee; }
|
|
|
|
scale { min-height: 24px; }
|
|
scale > contents > trough {
|
|
background-color: rgba(61, 55, 82, 0.95);
|
|
border: none; border-radius: 6px; min-height: 7px;
|
|
}
|
|
scale > contents > trough > highlight {
|
|
background-color: #92b9ea; border-radius: 6px;
|
|
}
|
|
scale > contents > trough > slider {
|
|
background-image: none;
|
|
background-color: #e9eefb;
|
|
border: 2px solid #0f0d17;
|
|
border-radius: 50%;
|
|
min-width: 14px; min-height: 14px;
|
|
margin: -6px;
|
|
}
|
|
scrollbar { background-color: transparent; }
|
|
scrollbar slider { background-color: rgba(146, 185, 234, 0.25); border-radius: 8px; }
|
|
"""
|
|
|
|
ICON = {
|
|
"net_eth": "", "net_wifi": "", "net_off": "",
|
|
"bt_on": "", "bt_off": "", "bt_conn": "",
|
|
"vol": "", "vol_mute": "", "mic": "", "mic_mute": "",
|
|
"dnd_on": "", "dnd_off": "", "particles": "", "wall": "",
|
|
"lock": "", "logout": "", "suspend": "", "reboot": "", "power": "",
|
|
"back": "", "refresh": "", "vpn_on": "", "vpn_off": "",
|
|
"clip": "", "shot": "", "bar": "",
|
|
}
|
|
|
|
|
|
def sh(cmd, timeout=6):
|
|
"""Run a command, return stdout ('' on any failure)."""
|
|
try:
|
|
return subprocess.run(cmd, shell=True, capture_output=True, text=True,
|
|
timeout=timeout).stdout.strip()
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def spawn(cmd):
|
|
subprocess.Popen(cmd, shell=True, start_new_session=True,
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
|
|
|
|
def has(binary):
|
|
return shutil.which(binary) is not None
|
|
|
|
|
|
def notify(title, body=""):
|
|
spawn(f'notify-send -a "Control" {GLib.shell_quote(title)} {GLib.shell_quote(body)}')
|
|
|
|
|
|
# ------------------------------------------------------------------ backends
|
|
def vol_get():
|
|
out = sh("wpctl get-volume @DEFAULT_AUDIO_SINK@")
|
|
m = re.search(r"([\d.]+)", out)
|
|
return (round(float(m.group(1)) * 100) if m else 0, "MUTED" in out)
|
|
|
|
|
|
def mic_get():
|
|
out = sh("wpctl get-volume @DEFAULT_AUDIO_SOURCE@")
|
|
m = re.search(r"([\d.]+)", out)
|
|
return (round(float(m.group(1)) * 100) if m else 0, "MUTED" in out)
|
|
|
|
|
|
def wp_ids(kind):
|
|
"""Map description -> WirePlumber node id for the Audio section.
|
|
|
|
pactl indices are a different namespace from PipeWire node ids, and
|
|
`wpctl set-default` only accepts the latter."""
|
|
out = sh("wpctl status")
|
|
audio = out.split("Audio", 1)[-1].split("Video", 1)[0]
|
|
ids, grab = {}, False
|
|
for line in audio.splitlines():
|
|
if re.search(r"\b(Sinks|Sources):", line):
|
|
grab = kind.rstrip("s").capitalize() + "s:" in line
|
|
continue
|
|
if re.search(r"\b(Devices|Filters|Streams|Sink endpoints|Source endpoints):", line):
|
|
grab = False
|
|
continue
|
|
if grab:
|
|
m = re.match(r"[^0-9*]*(\*?)\s*(\d+)\.\s+(.+?)(?:\s+\[vol.*)?$", line)
|
|
if m:
|
|
ids[m.group(3).strip()] = int(m.group(2))
|
|
return ids
|
|
|
|
|
|
def _pactl_devices(kind):
|
|
"""[(name, description, is_default, available, node_id)] for sinks/sources."""
|
|
default = sh(f"pactl get-default-{'sink' if kind == 'sinks' else 'source'}")
|
|
node_ids = wp_ids(kind)
|
|
out, items = sh(f"pactl list {kind}"), []
|
|
name = desc = None
|
|
ports, has_ports = [], False
|
|
blocks = re.split(r"\n(?=\w)", out)
|
|
for block in blocks:
|
|
m = re.search(r"^\s*Name:\s*(\S+)", block, re.M)
|
|
d = re.search(r"^\s*Description:\s*(.+)$", block, re.M)
|
|
if not m or not d:
|
|
continue
|
|
name, desc = m.group(1), d.group(1).strip()
|
|
port_lines = re.findall(r"^\s+\S+:.*\((?:type:.*?)?priority:.*?\)\s*$",
|
|
block, re.M)
|
|
has_ports = bool(port_lines)
|
|
# a device whose every port is unplugged cannot become the default
|
|
available = (not has_ports) or any("not available" not in p for p in port_lines)
|
|
items.append((name, desc, name == default, available,
|
|
node_ids.get(desc)))
|
|
return items
|
|
|
|
|
|
def sink_inputs():
|
|
"""Per-application streams: [(index, app, volume_pct, muted)]."""
|
|
out, cur, items = sh("pactl list sink-inputs"), {}, []
|
|
for line in out.splitlines():
|
|
st = line.strip()
|
|
if st.startswith("Sink Input #"):
|
|
if cur.get("idx"):
|
|
items.append((cur["idx"], cur.get("app", "?"),
|
|
cur.get("vol", 0), cur.get("mute", False)))
|
|
cur = {"idx": st.split("#")[1]}
|
|
elif st.startswith("Volume:") and "vol" not in cur:
|
|
m = re.search(r"(\d+)%", st)
|
|
cur["vol"] = int(m.group(1)) if m else 0
|
|
elif st.startswith("Mute:"):
|
|
cur["mute"] = st.endswith("yes")
|
|
elif "application.name" in st:
|
|
cur["app"] = st.split("=", 1)[1].strip().strip('"')
|
|
if cur.get("idx"):
|
|
items.append((cur["idx"], cur.get("app", "?"),
|
|
cur.get("vol", 0), cur.get("mute", False)))
|
|
return items
|
|
|
|
|
|
def net_state():
|
|
if not has("nmcli"):
|
|
return ICON["net_off"], "no NetworkManager", False
|
|
if "ethernet:connected" in sh("nmcli -t -f TYPE,STATE device status"):
|
|
return ICON["net_eth"], "Ethernet", True
|
|
if sh("nmcli -t -f WIFI radio") != "enabled":
|
|
return ICON["net_off"], "Wi-Fi off", False
|
|
for line in sh("nmcli -t -f NAME,TYPE connection show --active").splitlines():
|
|
if "wireless" in line:
|
|
return ICON["net_wifi"], line.split(":")[0], True
|
|
return ICON["net_wifi"], "not connected", False
|
|
|
|
|
|
def bt_state():
|
|
if not has("bluetoothctl"):
|
|
return ICON["bt_off"], "unavailable", False
|
|
if "Powered: yes" not in sh("bluetoothctl show"):
|
|
return ICON["bt_off"], "Off", False
|
|
n = len([l for l in sh("bluetoothctl devices Connected").splitlines() if l.startswith("Device")])
|
|
return (ICON["bt_conn"], f"{n} connected", True) if n else (ICON["bt_on"], "On", True)
|
|
|
|
|
|
def dnd_on():
|
|
return sh("dunstctl is-paused") == "true"
|
|
|
|
|
|
def vpn_state():
|
|
ifs = [l.split(":")[0].split("@")[0] for l in
|
|
sh("ip -br link show type wireguard").splitlines() if l.strip()]
|
|
ifs = [l.split()[0] for l in sh("ip -br link show type wireguard").splitlines() if l.strip()]
|
|
return (ICON["vpn_on"], ", ".join(ifs), True) if ifs else (ICON["vpn_off"], "off", False)
|
|
|
|
|
|
def wg_units():
|
|
out = sh("systemctl list-units --all --type=service --no-legend 'wg-quick@*'")
|
|
return [l.split()[0].replace("wg-quick@", "").replace(".service", "")
|
|
for l in out.splitlines() if l.strip()]
|
|
|
|
|
|
def particles_on():
|
|
return "running" in sh(f"{os.path.expanduser('~/.local/bin/particles')} --status || particles --status")
|
|
|
|
|
|
# --------------------------------------------------------------------- panel
|
|
class Panel(Gtk.Window):
|
|
def __init__(self):
|
|
super().__init__(title="Control Panel")
|
|
self.set_name("control-panel")
|
|
self.get_style_context().add_class("panel")
|
|
self.set_default_size(420, -1)
|
|
self.set_position(Gtk.WindowPosition.CENTER_ALWAYS)
|
|
self.set_decorated(False)
|
|
self.set_skip_taskbar_hint(True)
|
|
self.set_type_hint(Gdk.WindowTypeHint.DIALOG)
|
|
self.set_keep_above(True)
|
|
|
|
# An RGBA visual is all that is needed for the css background to blend.
|
|
# set_app_paintable(True) would make GTK skip drawing it altogether,
|
|
# which is what left the panel fully see-through.
|
|
visual = self.get_screen().get_rgba_visual()
|
|
if visual:
|
|
self.set_visual(visual)
|
|
# the window must not paint, or it paints opaque over the rgba visual
|
|
self.set_app_paintable(True)
|
|
|
|
self.keymap = {}
|
|
self.stack = Gtk.Stack()
|
|
self.stack.set_transition_type(Gtk.StackTransitionType.SLIDE_LEFT_RIGHT)
|
|
self.stack.set_transition_duration(140)
|
|
self.stack.set_vhomogeneous(False)
|
|
self.stack.set_hhomogeneous(True)
|
|
|
|
self.stack.add_named(self.build_main(), "main")
|
|
self.stack.add_named(self.build_list_view("Bluetooth", self.fill_bt), "bt")
|
|
self.stack.add_named(self.build_list_view("Network", self.fill_net), "net")
|
|
self.stack.add_named(self.build_audio_view(), "audio")
|
|
|
|
frame = Gtk.Box(orientation=Gtk.Orientation.VERTICAL)
|
|
frame.get_style_context().add_class("frame")
|
|
for setter in ("set_margin_top", "set_margin_bottom",
|
|
"set_margin_start", "set_margin_end"):
|
|
getattr(frame, setter)(12) # transparent gutter outside the card
|
|
frame.add(self.stack)
|
|
self.add(frame)
|
|
|
|
self.connect("key-press-event", self.on_key)
|
|
self.connect("map-event", self.on_map)
|
|
self.connect("destroy", Gtk.main_quit)
|
|
|
|
self.refresh()
|
|
GLib.timeout_add_seconds(3, self.refresh_tick)
|
|
|
|
# ---------------------------------------------------------- construction
|
|
def build_main(self):
|
|
box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=16)
|
|
|
|
head = Gtk.Box(spacing=10)
|
|
head.set_margin_bottom(2)
|
|
who = Gtk.Label(xalign=0)
|
|
who.set_markup(f"<b>{GLib.get_user_name()}</b>@{GLib.get_host_name()}")
|
|
who.get_style_context().add_class("title")
|
|
self.uptime = Gtk.Label(xalign=0)
|
|
self.uptime.get_style_context().add_class("subtle")
|
|
col = Gtk.Box(orientation=Gtk.Orientation.VERTICAL)
|
|
col.add(who)
|
|
col.add(self.uptime)
|
|
head.pack_start(col, True, True, 0)
|
|
box.add(head)
|
|
|
|
# quick toggles
|
|
grid = Gtk.Grid(column_spacing=10, row_spacing=10,
|
|
column_homogeneous=True)
|
|
self.t_net = self.tile(ICON["net_eth"], "Network", lambda *_: self.go("net"))
|
|
self.t_bt = self.tile(ICON["bt_on"], "Bluetooth", lambda *_: self.go("bt"))
|
|
self.t_vpn = self.tile(ICON["vpn_off"], "VPN", self.toggle_vpn)
|
|
self.t_dnd = self.tile(ICON["dnd_off"], "Notify", self.toggle_dnd)
|
|
self.t_par = self.tile(ICON["particles"], "Particles", self.toggle_particles)
|
|
self.t_wall = self.tile(ICON["wall"], "Wallpaper", lambda *_: self.shuffle_wall())
|
|
tiles = ((self.t_net, "n"), (self.t_bt, "b"), (self.t_vpn, "v"),
|
|
(self.t_dnd, "d"), (self.t_par, "p"), (self.t_wall, "w"))
|
|
for i, (w, k) in enumerate(tiles):
|
|
grid.attach(self.hint(w, k), i % 3, i // 3, 1, 1)
|
|
self.keymap[k] = (lambda _w=None, btn=w: btn.emit("clicked"))
|
|
box.add(grid)
|
|
|
|
# sliders
|
|
card = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=8)
|
|
card.get_style_context().add_class("card")
|
|
vol_row, self.vol_btn, self.vol_scale = self.slider_row(
|
|
ICON["vol"], self.on_vol, self.toggle_mute)
|
|
mic_row, self.mic_btn, self.mic_scale = self.slider_row(
|
|
ICON["mic"], self.on_mic, self.toggle_micmute)
|
|
card.add(vol_row)
|
|
card.add(mic_row)
|
|
box.add(card)
|
|
|
|
# power row
|
|
prow = Gtk.Box(spacing=8, homogeneous=True)
|
|
for icon, tip, act, danger, key in (
|
|
(ICON["lock"], "Lock", self.do_lock, False, "l"),
|
|
(ICON["logout"], "Log out", lambda *_: spawn("bspc quit"), False, "o"),
|
|
(ICON["suspend"], "Suspend",
|
|
lambda *_: (self.do_lock(), spawn("systemctl suspend")), False, "z"),
|
|
(ICON["reboot"], "Reboot", lambda *_: spawn("systemctl reboot"), True, "r"),
|
|
(ICON["power"], "Shut down", lambda *_: spawn("systemctl poweroff"), True, "q"),
|
|
):
|
|
b = Gtk.Button(label=icon)
|
|
b.set_tooltip_text(f"{tip} ({key})")
|
|
b.get_style_context().add_class("pw")
|
|
if danger:
|
|
b.get_style_context().add_class("danger")
|
|
b.connect("clicked", act)
|
|
self.keymap[key] = act
|
|
prow.add(self.hint(b, key))
|
|
box.add(prow)
|
|
|
|
extra = Gtk.Box(spacing=8, homogeneous=True)
|
|
for icon, tip, cmd, key in (
|
|
(ICON["vol"], "Audio devices & app volumes", "@audio", "a"),
|
|
(ICON["clip"], "Clipboard history",
|
|
"sh ~/.config/scripts/rofi-menus/clipboard-manager.sh", "c"),
|
|
(ICON["shot"], "Screenshot", "flameshot gui", "s"),
|
|
(ICON["bar"], "Toggle bar", "sh ~/.config/scripts/toggle-bar.sh", "t"),
|
|
):
|
|
b = Gtk.Button(label=icon)
|
|
b.set_tooltip_text(f"{tip} ({key})")
|
|
b.get_style_context().add_class("pw")
|
|
if cmd == "@audio":
|
|
act = lambda *_: self.go("audio")
|
|
else:
|
|
act = lambda _w=None, c=cmd: (spawn(c), self.close())
|
|
b.connect("clicked", act)
|
|
self.keymap[key] = act
|
|
extra.add(self.hint(b, key))
|
|
box.add(extra)
|
|
return box
|
|
|
|
def hint(self, widget, letter):
|
|
"""Wrap a widget so a small key hint sits in its bottom-right."""
|
|
ov = Gtk.Overlay()
|
|
ov.add(widget)
|
|
lab = Gtk.Label(label=letter)
|
|
lab.get_style_context().add_class("hint")
|
|
lab.set_halign(Gtk.Align.END)
|
|
lab.set_valign(Gtk.Align.END)
|
|
lab.set_margin_end(7)
|
|
lab.set_margin_bottom(5)
|
|
ov.add_overlay(lab)
|
|
ov.set_overlay_pass_through(lab, True)
|
|
ov.btn = widget
|
|
return ov
|
|
|
|
def tile(self, icon, label, cb):
|
|
b = Gtk.Button()
|
|
b.get_style_context().add_class("tile")
|
|
inner = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=3)
|
|
ic = Gtk.Label(label=icon)
|
|
ic.get_style_context().add_class("tile-icon")
|
|
tx = Gtk.Label(label=label)
|
|
tx.get_style_context().add_class("tile-name")
|
|
sub = Gtk.Label(label="")
|
|
sub.get_style_context().add_class("subtle")
|
|
inner.add(ic)
|
|
inner.add(tx)
|
|
inner.add(sub)
|
|
b.add(inner)
|
|
b.connect("clicked", cb)
|
|
b._icon, b._sub = ic, sub
|
|
return b
|
|
|
|
def slider_row(self, icon, on_change, on_click):
|
|
row = Gtk.Box(spacing=10)
|
|
btn = Gtk.Button(label=icon)
|
|
btn.get_style_context().add_class("volbtn")
|
|
btn.connect("clicked", on_click)
|
|
scale = Gtk.Scale.new_with_range(Gtk.Orientation.HORIZONTAL, 0, 100, 1)
|
|
scale.set_draw_value(False)
|
|
scale.set_hexpand(True)
|
|
scale.connect("value-changed", on_change)
|
|
pct = Gtk.Label(label="")
|
|
pct.get_style_context().add_class("subtle")
|
|
row.pack_start(btn, False, False, 0)
|
|
row.pack_start(scale, True, True, 0)
|
|
row.pack_start(pct, False, False, 0)
|
|
scale._pct = pct
|
|
return row, btn, scale
|
|
|
|
def build_list_view(self, title, filler):
|
|
outer = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=10)
|
|
head = Gtk.Box(spacing=8)
|
|
back = Gtk.Button(label=f"{ICON['back']} {title}")
|
|
back.get_style_context().add_class("rowbtn")
|
|
back.connect("clicked", lambda *_: self.go("main"))
|
|
head.pack_start(back, True, True, 0)
|
|
rescan = Gtk.Button(label=ICON["refresh"])
|
|
rescan.get_style_context().add_class("rowbtn")
|
|
rescan.connect("clicked", lambda *_: self.rescan(title))
|
|
head.pack_end(rescan, False, False, 0)
|
|
outer.add(head)
|
|
|
|
scroller = Gtk.ScrolledWindow()
|
|
scroller.set_policy(Gtk.PolicyType.NEVER, Gtk.PolicyType.AUTOMATIC)
|
|
scroller.set_min_content_height(120)
|
|
scroller.set_max_content_height(520)
|
|
scroller.set_propagate_natural_height(True)
|
|
listbox = Gtk.ListBox()
|
|
listbox.set_selection_mode(Gtk.SelectionMode.NONE)
|
|
listbox.get_style_context().add_class("card")
|
|
scroller.add(listbox)
|
|
outer.add(scroller)
|
|
setattr(self, f"list_{title.lower()}", listbox)
|
|
setattr(self, f"fill_{title.lower()}", filler)
|
|
return outer
|
|
|
|
def build_audio_view(self):
|
|
outer = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=16)
|
|
head = Gtk.Box(spacing=8)
|
|
back = Gtk.Button(label=f"{ICON['back']} Audio")
|
|
back.get_style_context().add_class("rowbtn")
|
|
back.connect("clicked", lambda *_: self.go("main"))
|
|
head.pack_start(back, True, True, 0)
|
|
outer.add(head)
|
|
# laid out like the home page: plain stacked cards, nothing scrolls
|
|
self.audio_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=16)
|
|
outer.add(self.audio_box)
|
|
return outer
|
|
|
|
def section(self, text):
|
|
lbl = Gtk.Label(label=text.upper(), xalign=0)
|
|
lbl.get_style_context().add_class("section")
|
|
lbl.set_margin_top(6)
|
|
lbl.set_margin_start(4)
|
|
lbl.set_margin_bottom(2)
|
|
return lbl
|
|
|
|
def device_row(self, name, desc, active, kind, available=True, node_id=None):
|
|
b = Gtk.Button()
|
|
b.get_style_context().add_class("rowbtn")
|
|
if active:
|
|
b.get_style_context().add_class("chosen")
|
|
h = Gtk.Box(spacing=10)
|
|
mark = Gtk.Label(label="" if active else "")
|
|
mark.get_style_context().add_class("icon")
|
|
h.add(mark)
|
|
t = Gtk.Label(label=desc, xalign=0)
|
|
t.set_ellipsize(3)
|
|
h.pack_start(t, True, True, 0)
|
|
if not available:
|
|
# a right-aligned tag, so ellipsis on a long name cannot eat it
|
|
tag = Gtk.Label(label="unplugged")
|
|
tag.get_style_context().add_class("subtle")
|
|
h.pack_end(tag, False, False, 0)
|
|
b.add(h)
|
|
if not available:
|
|
b.get_style_context().add_class("unavail")
|
|
b.connect("clicked",
|
|
lambda *_: self.set_device(kind, name, node_id, available))
|
|
return b
|
|
|
|
def app_row(self, idx, app, vol, muted, show_label=True, icon=None):
|
|
wp = str(idx).startswith("@")
|
|
card = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=2)
|
|
lbl = Gtk.Label(label=app, xalign=0)
|
|
lbl.get_style_context().add_class("subtle")
|
|
row = Gtk.Box(spacing=10)
|
|
base = icon or ICON["vol"]
|
|
muted_icon = {ICON["vol"]: ICON["vol_mute"],
|
|
ICON["mic"]: ICON["mic_mute"]}.get(base, ICON["vol_mute"])
|
|
mb = Gtk.Button(label=muted_icon if muted else base)
|
|
mb.get_style_context().add_class("volbtn")
|
|
mb.connect("clicked", lambda *_: (
|
|
sh(f"wpctl set-mute {idx} toggle" if wp
|
|
else f"pactl set-sink-input-mute {idx} toggle"),
|
|
self.fill_audio(), self.refresh()))
|
|
sc = Gtk.Scale.new_with_range(Gtk.Orientation.HORIZONTAL, 0, 130, 1)
|
|
sc.set_draw_value(False)
|
|
sc.set_hexpand(True)
|
|
sc.set_value(vol)
|
|
pct = Gtk.Label(label=f"{vol}%")
|
|
pct.get_style_context().add_class("subtle")
|
|
sc.connect("value-changed", lambda w: (
|
|
pct.set_text(f"{int(w.get_value())}%"),
|
|
spawn(f"wpctl set-volume {idx} {int(w.get_value())}%" if wp
|
|
else f"pactl set-sink-input-volume {idx} {int(w.get_value())}%")))
|
|
row.pack_start(mb, False, False, 0)
|
|
row.pack_start(sc, True, True, 0)
|
|
row.pack_end(pct, False, False, 0)
|
|
if show_label:
|
|
card.add(lbl)
|
|
card.add(row)
|
|
return card
|
|
|
|
def set_device(self, kind, name, node_id=None, available=True):
|
|
label = "Output" if kind == "sink" else "Input"
|
|
if not available:
|
|
notify("Audio", f"{label} unavailable - nothing plugged into that jack")
|
|
return
|
|
before = sh(f"pactl get-default-{kind}")
|
|
# pactl set-default-* is a no-op under PipeWire; WirePlumber owns the
|
|
# default and only honours wpctl with a node id
|
|
if node_id is not None:
|
|
sh(f"wpctl set-default {node_id}")
|
|
else:
|
|
sh(f"pactl set-default-{kind} {name}")
|
|
GLib.usleep(250000)
|
|
after = sh(f"pactl get-default-{kind}")
|
|
if after == before and before != name:
|
|
notify("Audio", f"{label} did not switch",
|
|
"WirePlumber refused it, usually an unplugged port")
|
|
else:
|
|
if kind == "sink":
|
|
for idx, *_ in sink_inputs(): # move streams to the new output
|
|
sh(f"pactl move-sink-input {idx} {name}")
|
|
notify("Audio", f"{label} switched")
|
|
self.fill_audio()
|
|
self.refresh()
|
|
|
|
def fill_audio(self):
|
|
for c in self.audio_box.get_children():
|
|
self.audio_box.remove(c)
|
|
|
|
card = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=10)
|
|
card.get_style_context().add_class("card")
|
|
v, vm = vol_get()
|
|
m, mm = mic_get()
|
|
card.add(self.app_row("@DEFAULT_AUDIO_SINK@", "", v, vm,
|
|
show_label=False, icon=ICON["vol"]))
|
|
card.add(self.app_row("@DEFAULT_AUDIO_SOURCE@", "", m, mm,
|
|
show_label=False, icon=ICON["mic"]))
|
|
self.audio_box.add(card)
|
|
|
|
self.audio_box.add(self.section("Output"))
|
|
card = Gtk.Box(orientation=Gtk.Orientation.VERTICAL)
|
|
card.get_style_context().add_class("card")
|
|
for name, desc, active, avail, nid in _pactl_devices("sinks"):
|
|
card.add(self.device_row(name, desc, active, "sink", avail, nid))
|
|
self.audio_box.add(card)
|
|
|
|
self.audio_box.add(self.section("Input"))
|
|
card = Gtk.Box(orientation=Gtk.Orientation.VERTICAL)
|
|
card.get_style_context().add_class("card")
|
|
for name, desc, active, avail, nid in _pactl_devices("sources"):
|
|
if ".monitor" in name:
|
|
continue
|
|
card.add(self.device_row(name, desc, active, "source", avail, nid))
|
|
self.audio_box.add(card)
|
|
|
|
apps = sink_inputs()
|
|
if apps:
|
|
self.audio_box.add(self.section("Applications"))
|
|
card = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=10)
|
|
card.get_style_context().add_class("card")
|
|
for idx, app, vol, muted in apps:
|
|
card.add(self.app_row(idx, app, vol, muted))
|
|
if len(apps) > 4: # only a long list needs to scroll
|
|
sc = Gtk.ScrolledWindow()
|
|
sc.set_policy(Gtk.PolicyType.NEVER, Gtk.PolicyType.AUTOMATIC)
|
|
sc.set_max_content_height(260)
|
|
sc.set_propagate_natural_height(True)
|
|
sc.add(card)
|
|
self.audio_box.add(sc)
|
|
else:
|
|
self.audio_box.add(card)
|
|
self.audio_box.show_all()
|
|
|
|
def on_map(self, *_):
|
|
"""Own the seat the way rofi does: keyboard and pointer both, so the
|
|
panel keeps focus wherever the cursor is or whatever gets clicked."""
|
|
self.present()
|
|
seat = Gdk.Display.get_default().get_default_seat()
|
|
gdkwin = self.get_window()
|
|
if seat and gdkwin:
|
|
seat.grab(gdkwin, Gdk.SeatCapabilities.ALL, True,
|
|
None, None, None, None)
|
|
return False
|
|
|
|
def release_grab(self):
|
|
seat = Gdk.Display.get_default().get_default_seat()
|
|
if seat:
|
|
seat.ungrab()
|
|
|
|
def close(self, *_):
|
|
self.release_grab()
|
|
super().close()
|
|
|
|
# -------------------------------------------------------------- actions
|
|
def go(self, name):
|
|
if name != "main":
|
|
{"bt": self.fill_bt, "net": self.fill_net, "audio": self.fill_audio}[name]()
|
|
self.stack.set_visible_child_name(name)
|
|
GLib.idle_add(lambda: (self.resize(1, 1), False)[1])
|
|
|
|
def on_key(self, _w, ev):
|
|
if ev.keyval == Gdk.KEY_Escape:
|
|
if self.stack.get_visible_child_name() != "main":
|
|
self.go("main") # step back, no relaunch
|
|
else:
|
|
self.close()
|
|
return True
|
|
if self.stack.get_visible_child_name() != "main":
|
|
return False
|
|
name = Gdk.keyval_name(ev.keyval) or ""
|
|
if name in ("m", "M"):
|
|
self.toggle_mute()
|
|
return True
|
|
if name in ("i", "I"):
|
|
self.toggle_micmute()
|
|
return True
|
|
if name in ("Left", "Right", "Up", "Down"):
|
|
step = -5 if name in ("Left", "Down") else 5
|
|
spawn(f"wpctl set-volume @DEFAULT_AUDIO_SINK@ "
|
|
f"{abs(step)}%{'-' if step < 0 else '+'} -l 1.3")
|
|
GLib.timeout_add(180, lambda: (self.refresh(), False)[1])
|
|
return True
|
|
act = self.keymap.get(name.lower())
|
|
if act:
|
|
act()
|
|
return True
|
|
return False
|
|
|
|
def on_vol(self, scale):
|
|
v = int(scale.get_value())
|
|
scale._pct.set_text(f"{v}%")
|
|
if not self._loading:
|
|
spawn(f"wpctl set-volume @DEFAULT_AUDIO_SINK@ {v}%")
|
|
|
|
def on_mic(self, scale):
|
|
v = int(scale.get_value())
|
|
scale._pct.set_text(f"{v}%")
|
|
if not self._loading:
|
|
spawn(f"wpctl set-volume @DEFAULT_AUDIO_SOURCE@ {v}%")
|
|
|
|
def toggle_mute(self, *_):
|
|
sh("wpctl set-mute @DEFAULT_AUDIO_SINK@ toggle")
|
|
self.refresh()
|
|
|
|
def toggle_micmute(self, *_):
|
|
sh("wpctl set-mute @DEFAULT_AUDIO_SOURCE@ toggle")
|
|
self.refresh()
|
|
|
|
def toggle_dnd(self, *_):
|
|
sh(f"dunstctl set-paused {'false' if dnd_on() else 'true'}")
|
|
self.refresh()
|
|
|
|
def toggle_particles(self, *_):
|
|
spawn("particles || ~/.local/bin/particles")
|
|
GLib.timeout_add(700, lambda: (self.refresh(), False)[1])
|
|
|
|
def shuffle_wall(self):
|
|
spawn("sh ~/.config/scripts/set-wallpaper.sh")
|
|
notify("Wallpaper", "Shuffled")
|
|
|
|
def toggle_vpn(self, *_):
|
|
_, _, up = vpn_state()
|
|
if up:
|
|
for i in [l.split()[0] for l in
|
|
sh("ip -br link show type wireguard").splitlines() if l.strip()]:
|
|
spawn(f"systemctl stop wg-quick@{i}")
|
|
notify("VPN", "Disconnecting")
|
|
else:
|
|
units = wg_units()
|
|
if not units:
|
|
notify("VPN", "No WireGuard interfaces configured")
|
|
return
|
|
spawn(f"systemctl start wg-quick@{units[0]}")
|
|
notify("VPN", f"Connecting {units[0]}…")
|
|
GLib.timeout_add_seconds(2, lambda: (self.refresh(), False)[1])
|
|
|
|
def do_lock(self, *_):
|
|
for cmd in ("betterlockscreen -l blur", "i3lock -c 0f0d17", "xsecurelock", "slock"):
|
|
if has(cmd.split()[0]):
|
|
spawn(cmd)
|
|
return
|
|
notify("Lock", "No screen locker installed")
|
|
|
|
def rescan(self, title):
|
|
if title == "Bluetooth":
|
|
spawn("bluetoothctl --timeout 20 scan on")
|
|
notify("Bluetooth", "Scanning…")
|
|
GLib.timeout_add_seconds(3, lambda: (self.fill_bt(), False)[1])
|
|
else:
|
|
spawn("nmcli device wifi rescan")
|
|
GLib.timeout_add_seconds(3, lambda: (self.fill_net(), False)[1])
|
|
|
|
def row(self, listbox, icon, label, sub, cb):
|
|
r = Gtk.ListBoxRow()
|
|
b = Gtk.Button()
|
|
b.get_style_context().add_class("rowbtn")
|
|
h = Gtk.Box(spacing=12)
|
|
ic = Gtk.Label(label=icon)
|
|
ic.get_style_context().add_class("icon")
|
|
h.add(ic)
|
|
t = Gtk.Label(label=label, xalign=0)
|
|
h.pack_start(t, True, True, 0)
|
|
s = Gtk.Label(label=sub)
|
|
s.get_style_context().add_class("subtle")
|
|
h.pack_end(s, False, False, 0)
|
|
b.add(h)
|
|
b.connect("clicked", cb)
|
|
r.add(b)
|
|
listbox.add(r)
|
|
|
|
def fill_bt(self):
|
|
lb = self.list_bluetooth
|
|
for c in lb.get_children():
|
|
lb.remove(c)
|
|
powered = "Powered: yes" in sh("bluetoothctl show")
|
|
self.row(lb, ICON["bt_on"] if powered else ICON["bt_off"],
|
|
"Bluetooth", "on" if powered else "off",
|
|
lambda *_: (sh(f"bluetoothctl power {'off' if powered else 'on'}"),
|
|
self.fill_bt(), self.refresh()))
|
|
if powered:
|
|
for line in sh("bluetoothctl devices").splitlines():
|
|
parts = line.split(" ", 2)
|
|
if len(parts) < 3:
|
|
continue
|
|
mac, name = parts[1], parts[2]
|
|
info = sh(f"bluetoothctl info {mac}")
|
|
conn = "Connected: yes" in info
|
|
state = "connected" if conn else ("paired" if "Paired: yes" in info else "")
|
|
self.row(lb, ICON["bt_conn"] if conn else ICON["bt_on"], name, state,
|
|
lambda _w, m=mac, n=name, c=conn: self.bt_toggle(m, n, c))
|
|
lb.show_all()
|
|
|
|
def bt_toggle(self, mac, name, connected):
|
|
if connected:
|
|
spawn(f"bluetoothctl disconnect {mac}")
|
|
notify("Bluetooth", f"{name} disconnected")
|
|
else:
|
|
notify("Bluetooth", f"Connecting to {name}…")
|
|
spawn(f"bluetoothctl trust {mac} && bluetoothctl connect {mac}")
|
|
GLib.timeout_add_seconds(3, lambda: (self.fill_bt(), self.refresh(), False)[2])
|
|
|
|
def fill_net(self):
|
|
lb = self.list_network
|
|
for c in lb.get_children():
|
|
lb.remove(c)
|
|
has_wifi = any(l.endswith(":wifi") for l in
|
|
sh("nmcli -t -f DEVICE,TYPE device status").splitlines())
|
|
if has_wifi:
|
|
on = sh("nmcli -t -f WIFI radio") == "enabled"
|
|
self.row(lb, ICON["net_wifi"] if on else ICON["net_off"], "Wi-Fi",
|
|
"on" if on else "off",
|
|
lambda *_: (sh(f"nmcli radio wifi {'off' if on else 'on'}"),
|
|
self.fill_net(), self.refresh()))
|
|
if on:
|
|
seen = set()
|
|
for line in sh("nmcli -t -f IN-USE,SIGNAL,SSID device wifi list").splitlines():
|
|
p = line.split(":")
|
|
if len(p) < 3 or not p[2] or p[2] in seen:
|
|
continue
|
|
seen.add(p[2])
|
|
self.row(lb, "" if p[0] == "*" else ICON["net_wifi"], p[2], f"{p[1]}%",
|
|
lambda _w, s=p[2]: self.wifi_connect(s))
|
|
else:
|
|
active = sh("nmcli -t -f NAME connection show --active").splitlines()
|
|
for line in sh("nmcli -t -f NAME,TYPE connection show").splitlines():
|
|
p = line.split(":")
|
|
if len(p) < 2 or p[1] in ("bridge", "loopback"):
|
|
continue
|
|
up = p[0] in active
|
|
self.row(lb, ICON["net_eth"] if up else "", p[0], "up" if up else "",
|
|
lambda _w, n=p[0], u=up: self.net_toggle(n, u))
|
|
lb.show_all()
|
|
|
|
def wifi_connect(self, ssid):
|
|
spawn(f"nmcli connection up id {GLib.shell_quote(ssid)} || "
|
|
f"nmcli device wifi connect {GLib.shell_quote(ssid)}")
|
|
notify("Wi-Fi", f"Connecting to {ssid}…")
|
|
GLib.timeout_add_seconds(3, lambda: (self.fill_net(), self.refresh(), False)[2])
|
|
|
|
def net_toggle(self, name, up):
|
|
spawn(f"nmcli connection {'down' if up else 'up'} id {GLib.shell_quote(name)}")
|
|
GLib.timeout_add_seconds(2, lambda: (self.fill_net(), self.refresh(), False)[2])
|
|
|
|
# -------------------------------------------------------------- refresh
|
|
def refresh_tick(self):
|
|
self.refresh()
|
|
return True
|
|
|
|
def refresh(self):
|
|
self._loading = True
|
|
up = sh("uptime -p") or ""
|
|
self.uptime.set_text(up.replace("up ", "up "))
|
|
|
|
icon, sub, on = net_state()
|
|
self.t_net._icon.set_text(icon)
|
|
self.t_net._sub.set_text(sub)
|
|
self.style(self.t_net, on)
|
|
|
|
icon, sub, on = bt_state()
|
|
self.t_bt._icon.set_text(icon)
|
|
self.t_bt._sub.set_text(sub)
|
|
self.style(self.t_bt, on)
|
|
|
|
paused = dnd_on()
|
|
self.t_dnd._icon.set_text(ICON["dnd_on"] if paused else ICON["dnd_off"])
|
|
self.t_dnd._sub.set_text("silenced" if paused else "on")
|
|
self.style(self.t_dnd, paused)
|
|
|
|
icon, sub, on = vpn_state()
|
|
self.t_vpn._icon.set_text(icon)
|
|
self.t_vpn._sub.set_text(sub)
|
|
self.style(self.t_vpn, on)
|
|
|
|
pon = particles_on()
|
|
self.t_par._sub.set_text("on" if pon else "off")
|
|
self.style(self.t_par, pon)
|
|
|
|
v, vm = vol_get()
|
|
self.vol_scale.set_value(v)
|
|
self.vol_scale._pct.set_text(f"{v}%")
|
|
self.vol_btn.set_label(ICON["vol_mute"] if vm else ICON["vol"])
|
|
|
|
m, mm = mic_get()
|
|
self.mic_scale.set_value(m)
|
|
self.mic_scale._pct.set_text(f"{m}%")
|
|
self.mic_btn.set_label(ICON["mic_mute"] if mm else ICON["mic"])
|
|
|
|
self._loading = False
|
|
return True
|
|
|
|
@staticmethod
|
|
def style(widget, active):
|
|
ctx = widget.get_style_context()
|
|
(ctx.add_class if active else ctx.remove_class)("active")
|
|
|
|
|
|
PIDFILE = os.path.join(
|
|
os.environ.get("XDG_RUNTIME_DIR", "/tmp"), "sakura-control-panel.pid")
|
|
|
|
|
|
def toggle_off_if_running():
|
|
"""A pidfile, not pgrep: matching on a command line also matches the shell
|
|
that launched us. Returns True when an existing panel was closed."""
|
|
try:
|
|
with open(PIDFILE) as fh:
|
|
pid = int(fh.read().strip())
|
|
except (OSError, ValueError):
|
|
return False
|
|
if pid == os.getpid():
|
|
return False
|
|
# the pidfile survives SIGTERM, so confirm the pid is really us before
|
|
# signalling it - pids get recycled
|
|
try:
|
|
with open(f"/proc/{pid}/cmdline", "rb") as fh:
|
|
if b"control-panel.py" not in fh.read():
|
|
return False
|
|
except OSError:
|
|
return False
|
|
try:
|
|
os.kill(pid, 15)
|
|
return True
|
|
except ProcessLookupError:
|
|
return False
|
|
except OSError:
|
|
return False
|
|
|
|
|
|
def main():
|
|
if toggle_off_if_running():
|
|
return
|
|
try:
|
|
with open(PIDFILE, "w") as fh:
|
|
fh.write(str(os.getpid()))
|
|
except OSError:
|
|
pass
|
|
|
|
provider = Gtk.CssProvider()
|
|
provider.load_from_data(CSS)
|
|
# ~/.config/gtk-3.0/gtk.css loads at PRIORITY_USER (800) and styles plain
|
|
# `button`, which outranks this sheet at APPLICATION (600) and was undoing
|
|
# the tile/icon styling. Sit just above it.
|
|
Gtk.StyleContext.add_provider_for_screen(
|
|
Gdk.Screen.get_default(), provider,
|
|
Gtk.STYLE_PROVIDER_PRIORITY_USER + 1)
|
|
win = Panel()
|
|
win._loading = True
|
|
win.show_all()
|
|
win.refresh()
|
|
try:
|
|
Gtk.main()
|
|
finally:
|
|
try:
|
|
with open(PIDFILE) as fh:
|
|
if int(fh.read().strip()) == os.getpid():
|
|
os.unlink(PIDFILE)
|
|
except (OSError, ValueError):
|
|
pass
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|