437 lines
20 KiB
Python
437 lines
20 KiB
Python
"""Full-width taskbar strip: jump to any open window, plus an expand-over panel.
|
||
|
||
Compact: app-grouped icons; click focuses the window (single) or opens a pop-out of
|
||
instances (grouped). The ⤢ toggle collapses this compact strip and hands its panel
|
||
(exposed as `panel_widget`) to the menu window, which mounts it over the quad region
|
||
so it *covers* the 2x2 quads rather than pushing anything down. The panel holds:
|
||
* workspace/layout controls — pick the current workspace's layout (scrolling /
|
||
dwindle / master / monocle, enumerated from ~/.cache/astal-menu/layouts.json,
|
||
written by hypr/layouts) and, for directional layouts, its direction. Applied
|
||
live via `hyprctl eval 'layouts.set(ws, name, dir)'`.
|
||
* a per-window row list: [icon + title → focus/jump] [⇤ pull to this workspace].
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
|
||
import gi
|
||
|
||
gi.require_version("Gtk", "4.0")
|
||
gi.require_version("AstalApps", "0.1")
|
||
from gi.repository import AstalApps, Gtk # noqa: E402
|
||
|
||
from lib.proc import run_json, run_text
|
||
from paths import CACHE_DIR
|
||
|
||
_LAYOUTS_MANIFEST = CACHE_DIR / "layouts.json"
|
||
_COLUMNS_STATE = CACHE_DIR / "columns-state.json"
|
||
_LAYOUTS_STATE = CACHE_DIR / "layouts-state.json"
|
||
|
||
|
||
def _eval(lua: str) -> None:
|
||
run_text(["hyprctl", "eval", lua], lambda *_a: None)
|
||
|
||
|
||
def _dispatch(lua: str) -> None:
|
||
# In hyprlua, `hyprctl dispatch` evaluates its argument as Lua (the hl.dsp.* API).
|
||
run_text(["hyprctl", "dispatch", lua], lambda *_a: None)
|
||
|
||
|
||
class Taskbar(Gtk.Box):
|
||
def __init__(self, on_activate, on_toggle_panel=None):
|
||
super().__init__(orientation=Gtk.Orientation.VERTICAL, spacing=6)
|
||
self.add_css_class("taskbar")
|
||
self._on_activate = on_activate
|
||
self._on_toggle_panel = on_toggle_panel # window mounts the panel over the quads
|
||
self._apps = AstalApps.Apps()
|
||
self._wm_index = self._build_wm_index()
|
||
self._layouts = self._load_layouts()
|
||
self._active_ws = None
|
||
self._clients: list = []
|
||
self._dir_dds = {}
|
||
self._fit_sws = {}
|
||
|
||
# The workspace/window panel. It is NOT appended here: the menu window mounts
|
||
# `panel_widget` into the quad region so that expanding COLLAPSES this strip's
|
||
# body (below) and the panel COVERS the 2x2 quads instead of pushing anything.
|
||
self._panel = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=8)
|
||
self._panel.add_css_class("task-panel")
|
||
self._panel_scroll = Gtk.ScrolledWindow(
|
||
hscrollbar_policy=Gtk.PolicyType.NEVER, vexpand=True)
|
||
self._panel_scroll.set_child(self._panel)
|
||
|
||
header = Gtk.CenterBox()
|
||
title = Gtk.Label(label="Open windows", xalign=0.0)
|
||
title.add_css_class("section-title")
|
||
header.set_start_widget(title)
|
||
self._expand_btn = Gtk.ToggleButton(label="") # nf-fa-expand
|
||
self._expand_btn.add_css_class("quad-action")
|
||
self._expand_btn.set_tooltip_text("Workspace & window controls")
|
||
self._expand_btn.connect("toggled", lambda b: self._toggle_panel(b.get_active()))
|
||
header.set_end_widget(self._expand_btn)
|
||
self.append(header)
|
||
|
||
# compact icon strip (hidden while the panel is expanded)
|
||
self._row = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8)
|
||
self._row.add_css_class("taskbar-row")
|
||
self._strip = Gtk.ScrolledWindow(
|
||
vscrollbar_policy=Gtk.PolicyType.NEVER,
|
||
hscrollbar_policy=Gtk.PolicyType.AUTOMATIC)
|
||
self._strip.set_child(self._row)
|
||
self.append(self._strip)
|
||
|
||
# -- expand-over-the-quads panel --------------------------------------
|
||
@property
|
||
def panel_widget(self) -> Gtk.Widget:
|
||
"""The workspace/window panel body; the window mounts this over the quads."""
|
||
return self._panel_scroll
|
||
|
||
def _toggle_panel(self, active: bool) -> None:
|
||
# Collapse this strip's compact body; the window reveals/hides the panel that
|
||
# it has mounted over the quad region (via on_toggle_panel).
|
||
self._strip.set_visible(not active)
|
||
if self._on_toggle_panel:
|
||
self._on_toggle_panel(active)
|
||
if active:
|
||
self.refresh()
|
||
|
||
def collapse_panel(self) -> None:
|
||
"""Return to the compact strip (called on menu hide / from the panel's Back)."""
|
||
if self._expand_btn.get_active():
|
||
self._expand_btn.set_active(False) # fires toggled -> _toggle_panel(False)
|
||
else:
|
||
self._strip.set_visible(True)
|
||
|
||
# -- setup helpers -----------------------------------------------------
|
||
def _build_wm_index(self) -> dict:
|
||
idx = {}
|
||
for app in self._apps.get_list():
|
||
for key in (app.get_wm_class(), app.get_executable(), app.get_name()):
|
||
if key:
|
||
idx.setdefault(key.lower(), app)
|
||
return idx
|
||
|
||
def _icon_for(self, cls: str) -> str:
|
||
app = self._wm_index.get((cls or "").lower())
|
||
if app and app.get_icon_name():
|
||
return app.get_icon_name()
|
||
return (cls or "application-x-executable").lower()
|
||
|
||
@staticmethod
|
||
def _load_layouts() -> list:
|
||
try:
|
||
return json.loads(_LAYOUTS_MANIFEST.read_text())
|
||
except (FileNotFoundError, json.JSONDecodeError):
|
||
# fallback if hypr/layouts hasn't written the manifest yet
|
||
return [{"name": "scrolling", "label": "Scrolling", "directional": True,
|
||
"dirs": ["down", "up", "right", "left"], "default_dir": "down",
|
||
"fit_method": True},
|
||
{"name": "columns", "label": "Columns", "directional": True,
|
||
"dirs": ["right", "down"], "dir_labels": ["Left / Right", "Up / Down"],
|
||
"default_dir": "right", "fit_method": True},
|
||
{"name": "dwindle", "label": "Dwindle", "directional": False, "dirs": []},
|
||
{"name": "master", "label": "Master", "directional": False, "dirs": []},
|
||
{"name": "monocle", "label": "Monocle", "directional": False, "dirs": []}]
|
||
|
||
# -- populate ----------------------------------------------------------
|
||
def refresh(self) -> None:
|
||
run_json(["hyprctl", "clients", "-j"], self._on_clients)
|
||
run_json(["hyprctl", "activeworkspace", "-j"], self._on_ws)
|
||
|
||
def _on_ws(self, ok: bool, data) -> None:
|
||
if ok and isinstance(data, dict):
|
||
self._active_ws = data.get("id")
|
||
self._rebuild_panel()
|
||
|
||
def _on_clients(self, ok: bool, data) -> None:
|
||
self._clients = data if ok and isinstance(data, list) else []
|
||
self._rebuild_strip()
|
||
self._rebuild_panel()
|
||
|
||
# -- compact strip -----------------------------------------------------
|
||
def _rebuild_strip(self) -> None:
|
||
self._clear(self._row)
|
||
groups: dict[str, list] = {}
|
||
for w in self._clients:
|
||
if not w.get("mapped", True) or not w.get("class"):
|
||
continue
|
||
groups.setdefault(w["class"], []).append(w)
|
||
if not groups:
|
||
self._row.append(Gtk.Label(label="No open windows"))
|
||
return
|
||
for cls, wins in sorted(groups.items()):
|
||
self._row.append(self._group_button(cls, wins))
|
||
|
||
def _group_button(self, cls: str, wins: list) -> Gtk.Widget:
|
||
icon = Gtk.Image.new_from_icon_name(self._icon_for(cls))
|
||
icon.set_pixel_size(32)
|
||
if len(wins) == 1:
|
||
btn = Gtk.Button()
|
||
btn.add_css_class("task-tile")
|
||
btn.set_child(icon)
|
||
btn.set_tooltip_text(wins[0].get("title") or cls)
|
||
btn.connect("clicked", lambda *_a, w=wins[0]: self._focus(w))
|
||
return btn
|
||
btn = Gtk.MenuButton()
|
||
btn.add_css_class("task-tile")
|
||
box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=6)
|
||
box.append(icon)
|
||
badge = Gtk.Label(label=str(len(wins)))
|
||
badge.add_css_class("task-badge")
|
||
box.append(badge)
|
||
btn.set_child(box)
|
||
btn.set_tooltip_text(f"{cls} ({len(wins)})")
|
||
pop = Gtk.Popover()
|
||
pop.add_css_class("task-popover")
|
||
plist = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=4)
|
||
for w in wins:
|
||
item = Gtk.Button(label=w.get("title") or cls)
|
||
item.add_css_class("task-window")
|
||
item.connect("clicked", lambda *_a, ww=w: (pop.popdown(), self._focus(ww)))
|
||
plist.append(item)
|
||
pop.set_child(plist)
|
||
btn.set_popover(pop)
|
||
return btn
|
||
|
||
# -- pop-open panel: layout controls + window list ---------------------
|
||
def _rebuild_panel(self) -> None:
|
||
self._clear(self._panel)
|
||
|
||
# tabbed layout selector for the active workspace: a tab per layout, and a
|
||
# per-layout options page underneath that switches with the tab. Selecting a
|
||
# tab applies that layout to the currently focused workspace.
|
||
ws = self._active_ws
|
||
hdr = Gtk.Label(label=f"Workspace {ws if ws is not None else '?'} layout", xalign=0.0)
|
||
hdr.add_css_class("section-title")
|
||
self._panel.append(hdr)
|
||
|
||
self._dir_dds = {} # layout name -> (Gtk.DropDown, [dir values])
|
||
self._fit_sws = {} # layout name -> Gtk.Switch
|
||
self._layout_stack = Gtk.Stack()
|
||
for ly in self._layouts:
|
||
self._layout_stack.add_titled(self._layout_page(ly), ly["name"], ly["label"])
|
||
switcher = Gtk.StackSwitcher(stack=self._layout_stack)
|
||
switcher.add_css_class("net-switcher") # reuse the tab pill styling
|
||
self._panel.append(switcher)
|
||
self._panel.append(self._layout_stack)
|
||
self._layout_stack.connect("notify::visible-child-name", self._on_tab_switch)
|
||
self._sync_layout_controls()
|
||
|
||
self._panel.append(Gtk.Separator())
|
||
|
||
# per-window rows: [focus/jump] [pull here]
|
||
wins = [w for w in self._clients if w.get("mapped", True) and w.get("class")]
|
||
wins.sort(key=lambda w: (w.get("workspace", {}).get("id", 0),
|
||
(w.get("title") or w.get("class") or "").lower()))
|
||
if not wins:
|
||
self._panel.append(Gtk.Label(label="No open windows", xalign=0.0))
|
||
return
|
||
listing = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=4)
|
||
scroller = Gtk.ScrolledWindow(vexpand=True, hscrollbar_policy=Gtk.PolicyType.NEVER)
|
||
scroller.set_child(listing)
|
||
for w in wins:
|
||
listing.append(self._window_row(w))
|
||
self._panel.append(scroller)
|
||
|
||
def _window_row(self, w) -> Gtk.Widget:
|
||
row = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8)
|
||
row.add_css_class("task-row")
|
||
icon = Gtk.Image.new_from_icon_name(self._icon_for(w.get("class", "")))
|
||
icon.set_pixel_size(22)
|
||
wsid = w.get("workspace", {}).get("id")
|
||
title = (w.get("title") or w.get("class") or "?")
|
||
name = Gtk.Button(hexpand=True)
|
||
name.add_css_class("task-name")
|
||
lbl = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8)
|
||
lbl.append(icon)
|
||
wl = Gtk.Label(label=f"{title}", xalign=0.0, hexpand=True, ellipsize=3, max_width_chars=32)
|
||
lbl.append(wl)
|
||
lbl.append(Gtk.Label(label=f"ws {wsid}", xalign=1.0))
|
||
name.set_child(lbl)
|
||
name.set_tooltip_text("Jump to window")
|
||
name.connect("clicked", lambda *_a: self._focus(w))
|
||
row.append(name)
|
||
|
||
pull = Gtk.Button(label="⇤ here")
|
||
pull.add_css_class("quad-action")
|
||
pull.set_tooltip_text("Pull this window to the current workspace")
|
||
pull.connect("clicked", lambda *_a: self._pull(w))
|
||
row.append(pull)
|
||
return row
|
||
|
||
# -- per-layout options page -------------------------------------------
|
||
def _layout_page(self, ly: dict) -> Gtk.Widget:
|
||
page = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=6)
|
||
page.add_css_class("layout-page")
|
||
if ly.get("dirs"):
|
||
row = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8)
|
||
row.append(Gtk.Label(label="direction", xalign=0.0, hexpand=True))
|
||
# dir_labels (optional) are friendly labels shown in place of the raw
|
||
# scrolling:direction values (e.g. "Left / Right" for "right").
|
||
labels = ly.get("dir_labels") or ly["dirs"]
|
||
dd = Gtk.DropDown.new_from_strings(labels)
|
||
dd.connect("notify::selected", self._on_opt_change)
|
||
row.append(dd)
|
||
page.append(row)
|
||
self._dir_dds[ly["name"]] = (dd, list(ly["dirs"]))
|
||
if ly.get("fit_method"):
|
||
row = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8)
|
||
row.append(Gtk.Label(label="center-focused", xalign=0.0, hexpand=True))
|
||
sw = Gtk.Switch(valign=Gtk.Align.CENTER)
|
||
if ly["name"] == "columns":
|
||
sw.set_tooltip_text("Keep the focused window centred in its column as "
|
||
"it scrolls, instead of the minimal-movement default")
|
||
else:
|
||
sw.set_tooltip_text("Keep the focused column centred so the prev/next "
|
||
"columns stay on-screen and tappable")
|
||
sw.connect("state-set", lambda s, state, name=ly["name"]: self._on_fit_toggle(s, state, name))
|
||
row.append(sw)
|
||
page.append(row)
|
||
self._fit_sws[ly["name"]] = sw
|
||
if ly.get("stepper"): # Columns: (-)[N](+)
|
||
page.append(self._cols_stepper())
|
||
if not ly.get("dirs") and not ly.get("fit_method") and not ly.get("stepper"):
|
||
page.append(Gtk.Label(label="No adjustable options", xalign=0.0,
|
||
css_classes=["net-ip"]))
|
||
return page
|
||
|
||
# -- columns count stepper --------------------------------------------
|
||
def _cols_stepper(self) -> Gtk.Widget:
|
||
row = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8)
|
||
row.append(Gtk.Label(label="Columns", xalign=0.0, hexpand=True))
|
||
minus = Gtk.Button(label="−")
|
||
minus.add_css_class("quad-action")
|
||
self._cols_lbl = Gtk.Label(label=str(self._read_cols()))
|
||
self._cols_lbl.add_css_class("stepper-value")
|
||
plus = Gtk.Button(label="+")
|
||
plus.add_css_class("quad-action")
|
||
minus.connect("clicked", lambda *_a: self._step_cols(-1))
|
||
plus.connect("clicked", lambda *_a: self._step_cols(1))
|
||
row.append(minus)
|
||
row.append(self._cols_lbl)
|
||
row.append(plus)
|
||
return row
|
||
|
||
def _read_cols(self) -> int:
|
||
try:
|
||
data = json.loads(_COLUMNS_STATE.read_text())
|
||
return int(data.get(str(self._active_ws), {}).get("cols", 2))
|
||
except (FileNotFoundError, json.JSONDecodeError, ValueError, TypeError):
|
||
return 2
|
||
|
||
def _step_cols(self, delta: int) -> None:
|
||
try:
|
||
cur = int(self._cols_lbl.get_label())
|
||
except ValueError:
|
||
cur = self._read_cols()
|
||
new = max(1, cur + delta)
|
||
self._cols_lbl.set_label(str(new))
|
||
_eval(f'hl.dispatch(hl.dsp.layout("cols {"+1" if delta > 0 else "-1"}"))')
|
||
|
||
def _read_center(self) -> bool:
|
||
# columns.lua's own "center-focused" switch — global, published in the same
|
||
# cache file as the columns stepper's count (see columns.lua's publish()).
|
||
try:
|
||
data = json.loads(_COLUMNS_STATE.read_text())
|
||
return bool(data.get("_center", False))
|
||
except (FileNotFoundError, json.JSONDecodeError, ValueError, TypeError):
|
||
return False
|
||
|
||
# -- sync + handlers ---------------------------------------------------
|
||
def _sync_layout_controls(self) -> None:
|
||
# reflect the current per-ws layout / direction / focus_fit in the tabs+options
|
||
self._syncing = True
|
||
run_json(["hyprctl", "getoption", "general:layout", "-j"], self._apply_tab_sel)
|
||
run_json(["hyprctl", "getoption", "scrolling:direction", "-j"], self._apply_dir_sel)
|
||
run_json(["hyprctl", "getoption", "scrolling:focus_fit_method", "-j"], self._apply_fit_sel)
|
||
# columns' own center-focused state lives in columns-state.json, not a hyprctl
|
||
# option — read it synchronously rather than round-tripping a subprocess.
|
||
columns_sw = self._fit_sws.get("columns")
|
||
if columns_sw is not None:
|
||
columns_sw.set_active(self._read_center())
|
||
|
||
def _apply_tab_sel(self, ok, data) -> None:
|
||
cur = data.get("str") if ok and isinstance(data, dict) else None
|
||
# prefer the per-workspace layout recorded by hypr/layouts (layouts.set)
|
||
try:
|
||
state = json.loads(_LAYOUTS_STATE.read_text())
|
||
cur = state.get(str(self._active_ws), cur)
|
||
except (FileNotFoundError, json.JSONDecodeError):
|
||
pass
|
||
if cur and any(ly["name"] == cur for ly in self._layouts):
|
||
self._layout_stack.set_visible_child_name(cur)
|
||
self._syncing = False
|
||
|
||
def _apply_dir_sel(self, ok, data) -> None:
|
||
# scrolling:direction is global; reflect it in every directional layout's
|
||
# dropdown whose value set contains it.
|
||
cur = data.get("str") if ok and isinstance(data, dict) else None
|
||
self._syncing = True
|
||
for dd, values in self._dir_dds.values():
|
||
if cur in values:
|
||
dd.set_selected(values.index(cur))
|
||
self._syncing = False
|
||
|
||
def _apply_fit_sel(self, ok, data) -> None:
|
||
# scrolling:focus_fit_method only describes the scrolling layout's own switch;
|
||
# columns' switch is synced separately from columns-state.json (see
|
||
# _sync_layout_controls) since it isn't backed by a hyprctl option at all.
|
||
val = data.get("int") if ok and isinstance(data, dict) else 0
|
||
sw = self._fit_sws.get("scrolling")
|
||
if sw is not None:
|
||
self._syncing = True
|
||
sw.set_active(val == 1)
|
||
self._syncing = False
|
||
|
||
def _cur_dir(self) -> str:
|
||
entry = self._dir_dds.get(self._layout_stack.get_visible_child_name())
|
||
if not entry:
|
||
return ""
|
||
dd, values = entry
|
||
i = dd.get_selected()
|
||
return values[i] if 0 <= i < len(values) else ""
|
||
|
||
def _on_tab_switch(self, *_a) -> None:
|
||
if getattr(self, "_syncing", False) or self._active_ws is None:
|
||
return
|
||
name = self._layout_stack.get_visible_child_name()
|
||
if name:
|
||
_eval(f'layouts.set("{self._active_ws}", "{name}", "{self._cur_dir()}")')
|
||
|
||
def _on_opt_change(self, *_a) -> None:
|
||
if getattr(self, "_syncing", False) or self._active_ws is None:
|
||
return
|
||
name = self._layout_stack.get_visible_child_name()
|
||
if name:
|
||
_eval(f'layouts.set("{self._active_ws}", "{name}", "{self._cur_dir()}")')
|
||
|
||
def _on_fit_toggle(self, _sw, state, name) -> bool:
|
||
if not getattr(self, "_syncing", False):
|
||
if name == "columns":
|
||
_eval(f'hl.dispatch(hl.dsp.layout("center {"on" if state else "off"}"))')
|
||
else:
|
||
_eval(f"layouts.set_fit({1 if state else 0})")
|
||
return False
|
||
|
||
# -- window actions ----------------------------------------------------
|
||
def _focus(self, w) -> None:
|
||
addr = w.get("address")
|
||
if addr:
|
||
_dispatch(f'hl.dsp.focus({{ window = "address:{addr}" }})')
|
||
self._on_activate()
|
||
|
||
def _pull(self, w) -> None:
|
||
addr = w.get("address")
|
||
if addr and self._active_ws is not None:
|
||
_dispatch(f'hl.dsp.window.move({{ window = "address:{addr}", '
|
||
f'workspace = "{self._active_ws}" }})')
|
||
run_json(["hyprctl", "clients", "-j"], self._on_clients) # reflect the move
|
||
|
||
@staticmethod
|
||
def _clear(box: Gtk.Box) -> None:
|
||
child = box.get_first_child()
|
||
while child:
|
||
box.remove(child)
|
||
child = box.get_first_child()
|