kjumpd.py

#!/usr/bin/python3
"""
kjumpd — KDE run-or-raise daemon.

Reads ~/.config/kjump/apps.json. For each app, registers a global shortcut
via kglobalaccel. On press, asks KWin to raise a matching window; if no
window matches, launches the configured command.

Config schema (see apps.example.json):

    {
      "apps": [
        {
          "id":              "firefox",                       # unique
          "name":            "Firefox",                       # shown in KDE settings
          "command":         ["firefox"],                     # Popen args
          "resourceClasses": ["org.mozilla.firefox", "firefox"],
          "shortcut":        268435540                        # Qt key code, 0 = unbound
        }
      ]
    }

D-Bus surface (so the GUI / `kjump` CLI can poke us):

    org.kjump.Daemon  /Daemon  org.kjump.Daemon
        Reload()             re-read config and apply diff
        Trigger(query) -> s  fire an entry (matched by id or name) without a
                             key press; returns the resolved id, "" if no match
        List() -> s          JSON array of {id, name, shortcut} entries
        Quit()               exit cleanly

CLI:
    ./kjumpd.py              run the daemon in the foreground
    ./kjumpd.py --version    print version and exit
    ./kjumpd.py --install    set up kjump: install + enable the systemd
                             user unit (shortcuts then appear automatically
                             from apps.json once the daemon runs)
    ./kjumpd.py --uninstall  remove kjump completely: stop & remove the
                             systemd unit AND remove all kjump shortcuts
                             from KDE
    ./kjumpd.py --help       this message
"""

import json
import os
import signal
import subprocess
import sys
import tempfile
import time
from pathlib import Path

import dbus
import dbus.mainloop.glib
import dbus.service
from gi.repository import GLib


# ---- Constants ------------------------------------------------------------

# Bump in lockstep with kjump.py and kjump-config.py — single-file scripts,
# no shared module to centralise it (see CLAUDE.md conventions).
__version__ = "0.1.0"

# No hyphen: KDE rewrites '-' to '_' in D-Bus object paths, which is ugly
# in System Settings (the component label uses the path, not the friendly name).
COMPONENT      = "kjump"
# COMPONENT_DESC is what kglobalaccel stores as `_k_friendly_name` and what
# System Settings → Shortcuts displays as the section title. Capitalised for
# readability — matches the convention of "KWin", "Konsole", etc.
COMPONENT_DESC = "KJump"
BUS_NAME       = "org.kjump.Daemon"
OBJ_PATH       = "/Daemon"

CONFIG_HOME = Path(os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config"))
CONFIG_DIR  = CONFIG_HOME / "kjump"
CONFIG_FILE = CONFIG_DIR / "apps.json"

UNIT_NAME = "kjumpd.service"
UNIT_FILE = CONFIG_HOME / "systemd" / "user" / UNIT_NAME

UNIT_TEMPLATE = """\
[Unit]
Description=kjump run-or-raise daemon
After=plasma-workspace.target
PartOf=plasma-workspace.target

[Service]
# Type=dbus: systemd considers us started when we acquire BusName, and won't
# launch a second copy while the name is owned.
Type=dbus
BusName={bus_name}
ExecStart={exec_start}
Restart=on-failure
RestartSec=3

[Install]
WantedBy=plasma-workspace.target
"""

# Owner-driven setShortcut with NoAutoloading. We tried setForeignShortcut
# which seemed simpler but turned out to only update kglobalaccel's
# in-memory state without persisting to ~/.config/kglobalshortcutsrc, which
# is where KWin reads its key-grab list from. With NoAutoloading the
# disk file gets the proper [component] section + action lines and KWin
# actually grabs the key combos. If kga rejects an override (returns the
# old value), we unRegister + doRegister + setShortcut as a fallback.
NO_AUTOLOADING = 0x2


def log(msg):
    t = time.strftime("%H:%M:%S")
    print(f"[{t}] {msg}", flush=True)


# ---- Config ---------------------------------------------------------------

def load_config():
    """Read apps.json. Returns list of app dicts, or None if the file is invalid."""
    if not CONFIG_FILE.exists():
        CONFIG_DIR.mkdir(parents=True, exist_ok=True)
        CONFIG_FILE.write_text('{"apps": []}\n')
        log(f"created empty config at {CONFIG_FILE}")
        return []
    try:
        data = json.loads(CONFIG_FILE.read_text())
    except json.JSONDecodeError as e:
        log(f"ERROR: bad JSON in {CONFIG_FILE}: {e}")
        return None

    apps = data.get("apps", [])
    seen = set()
    for a in apps:
        for required in ("id", "name", "command", "resourceClasses"):
            if required not in a:
                log(f"ERROR: app missing field {required!r}: {a}")
                return None
        if a["id"] in seen:
            log(f"ERROR: duplicate app id {a['id']!r}")
            return None
        seen.add(a["id"])
        a.setdefault("shortcut", 0)
    return apps


# ---- Pure helpers (unit-tested; see tests/) -------------------------------

def resolve(apps, query):
    """Find an app by id or name within `apps` (an id->app mapping). Match
    precedence: exact id, then case-insensitive id, then case-insensitive
    name. Returns the app dict or None. Lets the CLI accept either the stable
    id (good for scripts) or the friendly name (good for typing)."""
    if query in apps:
        return apps[query]
    q = query.casefold()
    for app in apps.values():
        if app["id"].casefold() == q:
            return app
    for app in apps.values():
        if app["name"].casefold() == q:
            return app
    return None


def strip_zero_keys(keys):
    """kga returns [0] (a single zero) for 'no binding', not []. Strip the
    zero sentinels so 'cleared' is never mistaken for a real key. Returns a
    list[int]."""
    return [int(k) for k in keys if int(k) != 0]


def flatten_key_seqs(seqs):
    """Flatten the a(ai) payload of yourShortcutsChanged: an array of
    QKeySequence structs, each holding one array of ints (padded with zeros).
    Three levels down to a flat list[int]."""
    return [k for seq in seqs for sub in seq for k in sub]


# ---- The daemon -----------------------------------------------------------

class Daemon:
    def __init__(self):
        self.apps = {}    # id -> app dict (currently registered)
        self.bus = dbus.SessionBus()

        self.kga = dbus.Interface(
            self.bus.get_object("org.kde.kglobalaccel", "/kglobalaccel"),
            "org.kde.KGlobalAccel",
        )
        # introspect=False so we can choose the 2-arg loadScript overload.
        self.kwin = dbus.Interface(
            self.bus.get_object("org.kde.KWin", "/Scripting", introspect=False),
            "org.kde.kwin.Scripting",
        )

        # The component path is deterministic: /component/<componentUnique>
        # (with hyphens rewritten to underscores by KDE — we chose a hyphenless
        # name to avoid the rewrite). We subscribe even before the component
        # exists in kglobalaccel: D-Bus match rules don't require the target
        # path to exist at subscribe time, the bus daemon just stores the rule.
        self._component_path = f"/component/{COMPONENT}"
        self.bus.add_signal_receiver(
            self._on_shortcut_pressed,
            signal_name="globalShortcutPressed",
            dbus_interface="org.kde.kglobalaccel.Component",
            bus_name="org.kde.kglobalaccel",
            path=self._component_path,
        )
        log(f"subscribed to globalShortcutPressed at {self._component_path}")

        # Catch shortcut changes from anywhere — System Settings, kcmshell,
        # third-party tools that talk to kglobalaccel. Two signals to listen
        # to because kglobalaccel uses the singular form for owner-driven
        # setShortcut and the plural form for setForeignShortcut (which is
        # what System Settings calls).
        self.bus.add_signal_receiver(
            lambda aid, keys: self._on_shortcut_changed(aid, list(keys)),
            signal_name="yourShortcutGotChanged",
            dbus_interface="org.kde.KGlobalAccel",
            bus_name="org.kde.kglobalaccel",
            path="/kglobalaccel",
        )
        # Plural signal: newKeys is a(ai) — array of QKeySequence structs,
        # each struct holding one array of ints (up to 4, padded with zeros).
        # Three levels: array of struct of array of int. Flatten all the way.
        self.bus.add_signal_receiver(
            lambda aid, seqs: self._on_shortcut_changed(
                aid, flatten_key_seqs(seqs)
            ),
            signal_name="yourShortcutsChanged",
            dbus_interface="org.kde.KGlobalAccel",
            bus_name="org.kde.kglobalaccel",
            path="/kglobalaccel",
        )
        log("subscribed to yourShortcut[s]Changed at /kglobalaccel")

    def _existing_kga_actions(self):
        """Action IDs that kglobalaccel currently has registered to our
        component. Returns [] if the component doesn't exist yet (first run)."""
        # introspect=False avoids a noisy UnknownObject error from dbus-python
        # auto-introspecting a path kglobalaccel hasn't created yet.
        comp = dbus.Interface(
            self.bus.get_object("org.kde.kglobalaccel", self._component_path,
                                introspect=False),
            "org.kde.kglobalaccel.Component",
        )
        try:
            return list(comp.shortcutNames())
        except dbus.exceptions.DBusException as e:
            # First-run case: component doesn't exist yet. Any of these names
            # means the same thing in practice; treat as "no existing actions".
            quiet = ("NoSuchComponent", "UnknownObject", "ServiceUnknown")
            if not any(q in str(e) for q in quiet):
                log(f"  could not query existing shortcuts: {e}")
            return []

    def _on_shortcut_pressed(self, component, action_id, _ts):
        if component != COMPONENT:
            return
        app = self.apps.get(action_id)
        if not app:
            log(f"signal for unknown action {action_id!r}")
            return
        self._try_raise_or_launch(app)

    def _on_shortcut_changed(self, action_id, flat_keys):
        """Fired by kglobalaccel when a registered shortcut's binding changes
        (either by us, the owner, or by another process via setForeignShortcut
        — System Settings uses the latter). The two callers above hand us a
        flattened list of key codes; we keep only the first non-zero one
        because kjump uses single-chord shortcuts.

        Echoes from our own setShortcut calls match the in-memory value
        (which we update BEFORE setShortcut) and silently no-op."""
        if len(action_id) < 2:
            return
        if str(action_id[0]) != COMPONENT:
            return
        app = self.apps.get(str(action_id[1]))
        if not app:
            # Unknown action — orphan being cleaned up, or one we haven't
            # registered yet.
            return
        nonzero = strip_zero_keys(flat_keys)
        new_value = nonzero[0] if nonzero else 0
        if int(app.get("shortcut", 0)) == new_value:
            return
        log(f"shortcut for {app['id']} changed externally -> "
            f"{hex(new_value) if new_value else 'cleared'}")
        app["shortcut"] = new_value
        self._save_apps_json()

    def _save_apps_json(self):
        """Write current self.apps state to apps.json atomically."""
        config = {"apps": list(self.apps.values())}
        CONFIG_DIR.mkdir(parents=True, exist_ok=True)
        tmp = CONFIG_FILE.with_suffix(".json.tmp")
        tmp.write_text(json.dumps(config, indent=2) + "\n")
        tmp.replace(CONFIG_FILE)
        log(f"  apps.json updated")

    def resolve(self, query):
        """Match `query` against the currently-registered apps. Thin wrapper
        over the module-level resolve() so the matching logic is unit-testable
        without a live session bus."""
        return resolve(self.apps, query)

    def _try_raise_or_launch(self, app):
        log(f"trigger {app['id']}")
        # json.dumps gives us safely-escaped JS string literals.
        classes_js = ", ".join(json.dumps(c) for c in app["resourceClasses"])
        action_js = json.dumps(app["id"])
        js = f"""
            let classes = [{classes_js}];
            let raised = false;
            for (let w of workspace.stackingOrder) {{
                if (w.normalWindow
                    && w.activities.includes(workspace.currentActivity)
                    && classes.includes(w.resourceClass)) {{
                    workspace.activeWindow = w;
                    raised = true;
                    break;
                }}
            }}
            callDBus("{BUS_NAME}", "{OBJ_PATH}", "{BUS_NAME}",
                     raised ? "windowRaised" : "windowNotFound", {action_js});
        """

        fd, path = tempfile.mkstemp(suffix=".js", prefix="kjump-")
        os.write(fd, js.encode())
        os.close(fd)

        plugin = f"kjump-{app['id']}-{time.time_ns()}"
        try:
            self.kwin.loadScript(path, plugin, signature="ss")
            self.kwin.start()
        except Exception as e:
            log(f"  KWin loadScript failed: {e}")
            os.unlink(path)
            return

        def cleanup():
            try:
                self.kwin.unloadScript(plugin)
                os.unlink(path)
            except Exception:
                pass
            return False
        GLib.timeout_add(2000, cleanup)

    def _action_id(self, app):
        return [COMPONENT, app["id"], COMPONENT_DESC, app["name"]]

    def register(self, app, startup=False):
        """Register the action with kglobalaccel and apply its key binding.

        startup=True is for the initial sync at daemon launch: if kga already
        has a non-empty binding for this action that differs from JSON, we
        prefer kga (the user may have edited it via System Settings while
        the daemon was off — silently overwriting that would be surprising).
        Mutates `app` in place when this happens, so the caller can detect
        and persist the change.

        Returns True iff app["shortcut"] was reconciled toward kga.
        """
        aid = self._action_id(app)
        self.kga.doRegister(aid)

        modified = False
        if startup:
            # kga returns [0] (single zero) for "no binding", not []. Strip
            # zero sentinels before comparing to avoid treating "cleared" as
            # a real binding that wins over JSON.
            kga_keys = strip_zero_keys(self.kga.shortcut(aid))
            json_keys = [int(app["shortcut"])] if app["shortcut"] else []
            if kga_keys and kga_keys != json_keys:
                log(f"  {app['id']}: kga has {hex(kga_keys[0])}, JSON had "
                    f"{hex(json_keys[0]) if json_keys else 'none'}; preferring kga")
                app["shortcut"] = kga_keys[0]
                modified = True

        # Update in-memory BEFORE setShortcut. The setShortcut call typically
        # fires yourShortcutGotChanged with the new value; if self.apps[id]
        # already matches, our handler treats it as an echo and writes nothing.
        self.apps[app["id"]] = app

        keys = [int(app["shortcut"])] if app["shortcut"] else []
        self._apply_binding(aid, keys, app)
        return modified

    def _apply_binding(self, aid, keys, app):
        """Push `keys` to kga's stored binding for action `aid`. Returns True
        on success. Detects key conflicts up front so a failed attempt never
        wipes a working binding."""
        if not keys:
            self.kga.setShortcut(aid, [], dbus.UInt32(NO_AUTOLOADING))
            log(f"  registered {app['id']} (no shortcut)")
            return True

        # Is this combo already grabbed by some OTHER action? Don't even try.
        try:
            owner = list(self.kga.action(int(keys[0])))
        except Exception:
            owner = []
        if owner and (len(owner) < 2 or
                      str(owner[0]) != COMPONENT or
                      str(owner[1]) != app["id"]):
            log(f"  WARNING: {hex(keys[0])} for {app['id']} is held by "
                f"{owner[0]}/{owner[1]}; not applied")
            return False

        bound_raw = self.kga.setShortcut(aid, keys, dbus.UInt32(NO_AUTOLOADING))
        bound_keys = strip_zero_keys(bound_raw)
        if bound_keys == keys:
            log(f"  registered {app['id']} -> {hex(keys[0])}")
            return True

        # First attempt didn't apply (NoAutoloading-only sometimes won't
        # override an already-stored binding). Fully reset and retry. This
        # is safe NOW because we already verified the key is free.
        log(f"  {app['id']}: setShortcut returned {bound_keys}; "
            f"resetting and retrying")
        try:
            self.kga.unRegister(aid)
        except Exception:
            pass
        self.kga.doRegister(aid)
        bound_raw = self.kga.setShortcut(aid, keys, dbus.UInt32(NO_AUTOLOADING))
        bound_keys = strip_zero_keys(bound_raw)
        if bound_keys == keys:
            log(f"  registered {app['id']} -> {hex(keys[0])} (after reset)")
            return True

        log(f"  WARNING: shortcut for {app['id']} -> {hex(keys[0])} "
            f"rejected; kga has {bound_keys or 'nothing'}")
        return False

    def unregister(self, app_id):
        app = self.apps.pop(app_id, None)
        if not app:
            return
        try:
            self.kga.unRegister(self._action_id(app))
            log(f"  unregistered {app_id}")
        except Exception as e:
            log(f"  unregister {app_id} failed: {e}")

    def apply(self, new_apps, startup=False):
        """Make kglobalaccel match new_apps:
          1. Unregister anything kglobalaccel knows about us that's not in
             new_apps (orphans from previous runs or removed entries).
          2. Register/update everything in new_apps (idempotent).

        startup=True: initial sync after daemon launch. For existing actions
            with a non-empty kga binding, kga wins over JSON (preserves user
            edits made via System Settings while daemon was off).
        startup=False: subsequent applies (Reload). JSON wins.
        """
        new_by_id = {a["id"]: a for a in new_apps}

        # Orphan cleanup. Uses the lowercase `unregister(component, action)`
        # variant which doesn't need the friendly names.
        for action_id in self._existing_kga_actions():
            if action_id in new_by_id:
                continue
            try:
                self.kga.unregister(COMPONENT, action_id)
                log(f"  cleaned up orphan {action_id}")
            except Exception as e:
                log(f"  orphan cleanup {action_id} failed: {e}")
            self.apps.pop(action_id, None)

        # Register / refresh.
        json_dirty = False
        for app in new_apps:
            old = self.apps.get(app["id"])
            if old == app:
                continue
            if self.register(app, startup=startup):
                json_dirty = True

        if json_dirty:
            self._save_apps_json()

    def shutdown(self):
        # Deliberately leaves shortcuts registered with kglobalaccel so that:
        #   - KDE Settings keeps showing them between daemon restarts
        #   - graceful and ungraceful shutdown have the same effect
        # Use --uninstall to remove all kjump entries from KDE.
        log("shutting down (registrations persist; use --uninstall to remove)")

    def uninstall(self):
        """Remove every kjump-component registration from kglobalaccel."""
        existing = self._existing_kga_actions()
        log(f"removing {len(existing)} kjump shortcut(s) from KDE")
        for action_id in existing:
            try:
                self.kga.unregister(COMPONENT, action_id)
                log(f"  removed {action_id}")
            except Exception as e:
                log(f"  remove {action_id} failed: {e}")


# ---- D-Bus service exposed by the daemon ---------------------------------

class KjumpService(dbus.service.Object):
    def __init__(self, daemon, conn, path):
        super().__init__(conn, path)
        self.daemon = daemon

    @dbus.service.method(BUS_NAME)
    def Reload(self):
        log("Reload()")
        new_apps = load_config()
        if new_apps is None:
            log("  config invalid; keeping current state")
            return
        self.daemon.apply(new_apps)

    @dbus.service.method(BUS_NAME)
    def Quit(self):
        log("Quit()")
        GLib.idle_add(_loop.quit)

    @dbus.service.method(BUS_NAME, in_signature="s", out_signature="s")
    def Trigger(self, query):
        """Fire an entry by id or name, as if its shortcut had been pressed.
        Returns the resolved entry id on success, or "" if nothing matched.
        This is what the `kjump` CLI calls, and the key-free invocation path
        for scripted testing."""
        query = str(query)  # arrives as dbus.String; log it cleanly
        log(f"Trigger({query!r})")
        app = self.daemon.resolve(query)
        if not app:
            log(f"  no entry matching {query!r}")
            return ""
        self.daemon._try_raise_or_launch(app)
        return app["id"]

    @dbus.service.method(BUS_NAME, out_signature="s")
    def List(self):
        """Return the currently-registered entries as a JSON array of
        {id, name, shortcut} objects. Backs `kjump --list`. Reflects live
        in-memory state, which the daemon keeps in sync with apps.json."""
        entries = [
            {"id": a["id"], "name": a["name"],
             "shortcut": int(a.get("shortcut", 0))}
            for a in self.daemon.apps.values()
        ]
        return json.dumps(entries)

    # --- callbacks invoked by the KWin one-shot script ---

    @dbus.service.method(BUS_NAME, in_signature="s")
    def windowRaised(self, action_id):
        log(f"  raised {action_id}")

    @dbus.service.method(BUS_NAME, in_signature="s")
    def windowNotFound(self, action_id):
        app = self.daemon.apps.get(action_id)
        if not app:
            return
        log(f"  no window for {action_id} -> launching {app['command']}")
        subprocess.Popen(
            app["command"],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            start_new_session=True,
        )


# ---- systemd unit install / uninstall ------------------------------------

def install_service():
    """Write the user unit, daemon-reload, then enable + start."""
    script = Path(__file__).resolve()
    if not os.access(script, os.X_OK):
        log(f"WARNING: {script} is not executable; chmod +x it first")
        return False

    UNIT_FILE.parent.mkdir(parents=True, exist_ok=True)
    UNIT_FILE.write_text(UNIT_TEMPLATE.format(
        bus_name=BUS_NAME,
        exec_start=script,
    ))
    log(f"wrote {UNIT_FILE}")

    for cmd in (
        ["systemctl", "--user", "daemon-reload"],
        ["systemctl", "--user", "enable", "--now", UNIT_NAME],
    ):
        log(f"  $ {' '.join(cmd)}")
        result = subprocess.run(cmd, capture_output=True, text=True)
        if result.returncode != 0:
            log(f"    FAILED ({result.returncode}): {result.stderr.strip()}")
            return False
    log(f"installed and started {UNIT_NAME}")
    return True


def uninstall_service():
    """Stop, disable, and remove the user unit. Idempotent."""
    if UNIT_FILE.exists():
        cmd = ["systemctl", "--user", "disable", "--now", UNIT_NAME]
        log(f"  $ {' '.join(cmd)}")
        subprocess.run(cmd, capture_output=True, text=True)  # ignore errors
        UNIT_FILE.unlink()
        log(f"removed {UNIT_FILE}")
        subprocess.run(["systemctl", "--user", "daemon-reload"],
                       capture_output=True)
    else:
        log(f"no unit at {UNIT_FILE}; nothing to remove")


def print_help():
    print(__doc__.strip())


# ---- Entrypoint ----------------------------------------------------------

_loop = None  # GLib mainloop ref so signal handlers can stop us

def main():
    global _loop

    if "--help" in sys.argv or "-h" in sys.argv:
        print_help()
        sys.exit(0)

    if "--version" in sys.argv or "-V" in sys.argv:
        print(f"kjumpd {__version__}")
        sys.exit(0)

    dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)

    if "--install" in sys.argv:
        sys.exit(0 if install_service() else 1)

    daemon = Daemon()

    if "--uninstall" in sys.argv:
        daemon.uninstall()        # remove kjump shortcuts from KDE
        uninstall_service()       # stop + remove the systemd unit
        sys.exit(0)

    # do_not_queue=True: refuse to start if another kjumpd is already running.
    # NB: we MUST keep a reference to the BusName object — dbus-python releases
    # the name as soon as the object is garbage-collected.
    try:
        _busname = dbus.service.BusName(BUS_NAME, daemon.bus, do_not_queue=True)
    except dbus.exceptions.NameExistsException:
        # Exit 0 (not 1): another kjumpd is already doing our job — that's not
        # a failure, and exit-1 would put systemd into a Restart=on-failure loop.
        log(f"another instance already owns {BUS_NAME}; exiting cleanly")
        sys.exit(0)
    daemon._busname = _busname  # keep alive for daemon lifetime
    daemon._service = KjumpService(daemon, daemon.bus, OBJ_PATH)
    log(f"published {BUS_NAME} at {OBJ_PATH}")

    apps = load_config()
    if apps is None:
        log("invalid config — exiting")
        sys.exit(1)
    daemon.apply(apps, startup=True)

    _loop = GLib.MainLoop()

    def stop(*_):
        _loop.quit()
    signal.signal(signal.SIGINT, stop)
    signal.signal(signal.SIGTERM, stop)

    log(f"running. {len(daemon.apps)} app(s) registered.")
    try:
        _loop.run()
    finally:
        daemon.shutdown()
        log("bye.")


if __name__ == "__main__":
    main()