From 42a4ffdee685703c40fb346e1084102250d38407 Mon Sep 17 00:00:00 2001 From: Makesesama Date: Sun, 4 Oct 2026 21:25:07 +0200 Subject: [PATCH] feat: dmenu-compatible picker --- sims/cli.py | 71 +++++++++++ sims/main.py | 10 +- sims/modules/launcher/__init__.py | 2 + sims/modules/launcher/dmenu.py | 199 ++++++++++++++++++++++++++++++ 4 files changed, 281 insertions(+), 1 deletion(-) create mode 100644 sims/modules/launcher/dmenu.py diff --git a/sims/cli.py b/sims/cli.py index 7b47542..526ffa2 100644 --- a/sims/cli.py +++ b/sims/cli.py @@ -3,9 +3,12 @@ from __future__ import annotations import argparse import json +import os import re +import socket import subprocess import sys +import tempfile from dataclasses import dataclass from typing import Callable @@ -104,6 +107,61 @@ def _cmd_invoke(ns: argparse.Namespace) -> None: invoke_action(ns.action) +def _cmd_dmenu(ns: argparse.Namespace) -> None: + """dmenu-compatible picker: lines on stdin, chosen line on stdout. + + Exits 1 when cancelled or when nothing was offered. Items travel via + a private request file instead of DBus arguments because dbus-send + splits array arguments on commas. + """ + lines = [line.rstrip("\r\n") for line in sys.stdin] + lines = [line for line in lines if line] + if not lines: + sys.exit(1) + + base = os.environ.get("XDG_RUNTIME_DIR") or None + with tempfile.TemporaryDirectory(prefix="sims-dmenu-", dir=base) as request_dir: + with open(os.path.join(request_dir, "request.json"), "w", encoding="utf-8") as f: + json.dump({"prompt": ns.prompt, "lines": lines}, f) + + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as server: + server.bind(os.path.join(request_dir, "reply.sock")) + server.listen(1) + server.settimeout(ns.timeout) + + out = _dbus_send( + f"{IFACE}.InvokeAction", + "string:open-dmenu", + f"array:string:{request_dir}", + ) + # InvokeAction replies (error: bool, message: str) with exit 0 + # even when the action does not exist (e.g. an older sims). + if re.search(r"^\s*boolean true\s*$", out, re.MULTILINE): + sys.stderr.write(f"sims-cli: open-dmenu failed:\n{out}") + sys.exit(1) + + try: + conn, _ = server.accept() + except socket.timeout: + sys.stderr.write("sims-cli: dmenu timed out\n") + sys.exit(1) + with conn: + conn.settimeout(5) + chunks = [] + try: + while chunk := conn.recv(4096): + chunks.append(chunk) + except OSError as e: + sys.stderr.write(f"sims-cli: dmenu reply failed: {e}\n") + sys.exit(1) + + choice = b"".join(chunks).decode("utf-8").rstrip("\n") + if not choice: + sys.exit(1) + sys.stdout.write(choice + "\n") + sys.stdout.flush() + + def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog="sims-cli", @@ -158,6 +216,19 @@ def build_parser() -> argparse.ArgumentParser: inv.add_argument("action", help="dbus action name, e.g. open-finder") inv.set_defaults(func=_cmd_invoke) + dm = sub.add_parser( + "dmenu", + help="dmenu-compatible picker (stdin lines -> chosen line on stdout)", + ) + dm.add_argument("-p", "--prompt", default="Select...", help="placeholder text") + dm.add_argument( + "--timeout", + type=float, + default=300.0, + help="seconds to wait for a choice before giving up (default: 300)", + ) + dm.set_defaults(func=_cmd_dmenu) + return parser diff --git a/sims/main.py b/sims/main.py index 4268daa..b267d37 100644 --- a/sims/main.py +++ b/sims/main.py @@ -22,6 +22,7 @@ from .modules.bar import StatusBar from .modules.window_fuzzy import FuzzyWindowFinder from .modules.launcher.apps import AppLauncher from .modules.launcher.clipboard import ClipboardMenu +from .modules.launcher.dmenu import DmenuMenu from .modules.launcher.notmuch_search import NotmuchSearchMenu from .modules.launcher.power import PowerMenu from .modules.launcher.screenrec import ScreenrecMenu @@ -55,6 +56,7 @@ clipboard_menu = ClipboardMenu() power_menu = PowerMenu(lock_command=POWER.get("lock_command", ["waylock"])) screenshot_menu = ScreenshotMenu() notmuch_search_menu = NotmuchSearchMenu() +dmenu_menu = DmenuMenu() screenrec_service: ScreenrecService | None = None screenrec_menu = None @@ -109,7 +111,7 @@ if ORG.get("enable", False): bar_windows = [] notmuch_widget = None -_app_windows = [dummy, finder, app_launcher, clipboard_menu, power_menu, screenshot_menu, notmuch_search_menu] +_app_windows = [dummy, finder, app_launcher, clipboard_menu, power_menu, screenshot_menu, notmuch_search_menu, dmenu_menu] if screenrec_menu is not None: _app_windows.append(screenrec_menu) if notification_toasts is not None: @@ -146,6 +148,12 @@ def open_screenshot_menu(): screenshot_menu.show() +@Application.action() +def open_dmenu(request_dir: str): + """Driven by `sims-cli dmenu`; see sims/modules/launcher/dmenu.py.""" + dmenu_menu.open_request(request_dir) + + @Application.action() def refresh_notmuch(): if notmuch_widget is not None: diff --git a/sims/modules/launcher/__init__.py b/sims/modules/launcher/__init__.py index a5b9659..39a84a3 100644 --- a/sims/modules/launcher/__init__.py +++ b/sims/modules/launcher/__init__.py @@ -1,7 +1,9 @@ from .base import FuzzyMenu, LauncherProvider, StaticAction, StaticActionProvider +from .dmenu import DmenuMenu from .windows import WindowProvider __all__ = [ + "DmenuMenu", "FuzzyMenu", "LauncherProvider", "StaticAction", diff --git a/sims/modules/launcher/dmenu.py b/sims/modules/launcher/dmenu.py new file mode 100644 index 0000000..6a27e84 --- /dev/null +++ b/sims/modules/launcher/dmenu.py @@ -0,0 +1,199 @@ +"""dmenu-compatible picker driven by ``sims-cli dmenu``. + +The CLI writes ``request.json`` (prompt + lines) into a private directory, +listens on ``reply.sock`` in that same directory and invokes the +``open-dmenu`` action with the directory path. The menu answers by +connecting to the socket and sending the chosen line verbatim (or nothing +when cancelled), so the CLI can print it on stdout like dmenu would. + +Lines produced by xdg-desktop-portal-wlr's screencast chooser +(``Monitor: `` / ``Window: (<identifier>)``) get +a nicer rendering; everything else is shown as plain text. +""" + +from __future__ import annotations + +import json +import os +import re +import socket +from dataclasses import dataclass + +from fabric.widgets.box import Box +from fabric.widgets.image import Image +from fabric.widgets.label import Label +from gi.repository import Gtk +from loguru import logger + +from sims.services.i3 import I3, I3MessageType + +from .base import FuzzyMenu +from .ranking import rank_items + +ICON_SIZE = 32 + +_MONITOR_RE = re.compile(r"^Monitor: (?P<name>\S+)(?: (?P<desc>.*))?$") +_WINDOW_RE = re.compile(r"^Window: (?P<title>.*) \((?P<ident>[^()]*)\)$") + +_WEIGHTS = {"title": 1.0, "detail": 0.6} + + +@dataclass +class DmenuItem: + line: str + kind: str # "monitor" | "window" | "plain" + title: str + detail: str = "" + icon_name: str | None = None + + +def _window_app_ids() -> dict[str, str]: + """Map window title -> app_id from fenster's tree (best effort).""" + reply = I3.send_command("", I3MessageType.GET_TREE) + if not (reply.is_ok and isinstance(reply.reply, dict)): + return {} + out: dict[str, str] = {} + stack = [reply.reply] + while stack: + node = stack.pop() + if node.get("type") == "con" and node.get("name") and node.get("app_id"): + out.setdefault(node["name"], node["app_id"]) + stack.extend(node.get("nodes", [])) + stack.extend(node.get("floating_nodes", [])) + return out + + +def _icon_for_app_id(app_id: str | None) -> str: + theme = Gtk.IconTheme.get_default() + if app_id and theme is not None: + for candidate in (app_id, app_id.lower(), app_id.rsplit(".", 1)[-1].lower()): + if theme.has_icon(candidate): + return candidate + return "application-x-executable" + + +def parse_lines(lines: list[str]) -> list[DmenuItem]: + app_ids: dict[str, str] | None = None + items: list[DmenuItem] = [] + for line in lines: + if m := _MONITOR_RE.match(line): + items.append( + DmenuItem( + line=line, + kind="monitor", + title=m["name"], + detail=m["desc"] or "", + icon_name="video-display", + ) + ) + elif m := _WINDOW_RE.match(line): + if app_ids is None: + try: + app_ids = _window_app_ids() + except Exception as e: # pragma: no cover - defensive + logger.warning(f"[dmenu] could not query window tree: {e}") + app_ids = {} + app_id = app_ids.get(m["title"]) + items.append( + DmenuItem( + line=line, + kind="window", + title=m["title"] or "(untitled)", + detail=app_id or "", + icon_name=_icon_for_app_id(app_id), + ) + ) + else: + items.append(DmenuItem(line=line, kind="plain", title=line)) + return items + + +class DmenuProvider: + def __init__(self): + self._items: list[DmenuItem] = [] + self.on_select = lambda _line: None + + def set_lines(self, lines: list[str]) -> None: + self._items = parse_lines(lines) + + def items(self) -> list[DmenuItem]: + return list(self._items) + + def filter(self, items: list[DmenuItem], query: str) -> list[DmenuItem]: + return rank_items( + items, + query, + lambda item: {"title": item.title, "detail": item.detail}, + weights=_WEIGHTS, + ) + + def render(self, item: DmenuItem) -> Gtk.Widget: + if item.kind == "plain": + return Box( + name="slot-box", + orientation="h", + children=[Label(label=item.title, h_align="start")], + ) + children: list[Gtk.Widget] = [] + if item.icon_name: + children.append( + Image(name="app-icon", icon_name=item.icon_name, icon_size=ICON_SIZE) + ) + text_box = Box(name="app-text", orientation="v", spacing=0) + text_box.add( + Label(label=item.title, name="app-name", h_align="start", ellipsization="end") + ) + kind = "Monitor" if item.kind == "monitor" else "Window" + detail = f"{kind} ยท {item.detail}" if item.detail else kind + text_box.add( + Label(label=detail, name="app-generic", h_align="start", ellipsization="end") + ) + children.append(text_box) + return Box(name="slot-box", orientation="h", spacing=10, children=children) + + def activate(self, item: DmenuItem) -> None: + self.on_select(item.line) + + +class DmenuMenu(FuzzyMenu): + def __init__(self, monitor: int = 0): + # Set before FuzzyMenu/Window init, which may call hide(). + self._reply_path: str | None = None + self._dmenu = DmenuProvider() + self._dmenu.on_select = self._finish + super().__init__( + provider=self._dmenu, + monitor=monitor, + placeholder="Select...", + window_name="dmenu", + max_results=10, + ) + + def open_request(self, request_dir: str) -> None: + with open(os.path.join(request_dir, "request.json"), encoding="utf-8") as f: + request = json.load(f) + # A new request supersedes any pending one. + self._finish(None) + self._reply_path = os.path.join(request_dir, "reply.sock") + self._dmenu.set_lines([str(line) for line in request.get("lines", [])]) + self.search_entry.set_placeholder_text(request.get("prompt") or "Select...") + self.show() + + def hide(self): + super().hide() + # Escape / Ctrl-G / activation all end up here; activation already + # replied, so this only fires for a real cancel. + self._finish(None) + + def _finish(self, line: str | None) -> None: + path, self._reply_path = self._reply_path, None + if path is None: + return + try: + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock: + sock.settimeout(1.0) + sock.connect(path) + if line is not None: + sock.sendall(line.encode("utf-8") + b"\n") + except OSError as e: + logger.warning(f"[dmenu] could not reply to {path}: {e}")