kadence/desktop/kadence_desktop/integrations/dbus_service.py

"""Kadence's own D-Bus API: `org.kadence.Desktop`.

This is what makes the app scriptable and bindable. KDE's System Settings →
Shortcuts → Custom Shortcuts can call a D-Bus method directly, so a global
hotkey for "start a pomodoro" needs no X11 key-grabbing hack:

    qdbus org.kadence.Desktop /Timer org.kadence.Timer.StartWork

Owning the name also gives single-instance behaviour for free: a second launch
sees the name is taken, calls `Raise()` on the first, and exits.

The verbs live on a **QDBusAbstractAdaptor**, not on the exported object
itself. Qt only honours the "D-Bus Interface" class info on an adaptor; export
a plain QObject and the interface is named after its class
(`local.kadence.TimerDBusService`), which is exactly the "No such interface"
every documented command then hits.
"""

from __future__ import annotations

import logging
from typing import Callable

from PyQt6.QtCore import QObject, pyqtClassInfo, pyqtSlot
from PyQt6.QtDBus import QDBusAbstractAdaptor, QDBusConnection

log = logging.getLogger(__name__)

SERVICE_NAME = "org.kadence.Desktop"
OBJECT_PATH = "/Timer"
INTERFACE = "org.kadence.Timer"


class TimerDBusService(QObject):
    """Holds the actions; the adaptor below is what the bus sees."""

    def __init__(self, actions: dict[str, Callable[[], object]], parent: QObject | None = None) -> None:
        super().__init__(parent)
        self._actions = actions
        self.adaptor = TimerAdaptor(self)

    def run(self, name: str) -> str:
        action = self._actions.get(name)
        if action is None:
            return f"unknown action: {name}"
        try:
            result = action()
            return str(result) if result is not None else "ok"
        except Exception as exc:  # pragma: no cover - never let a caller crash us
            log.exception("D-Bus action %s failed", name)
            return f"error: {exc}"


# Qt generates no introspection XML for a Python adaptor, and without it
# `busctl introspect` and KDE's D-Bus shortcut browser have nothing to list —
# so the interface describes itself. Every verb takes no arguments and returns
# one string.
VERBS = (
    "StartWork",
    "StartBreak",
    "Toggle",
    "Pause",
    "Resume",
    "Complete",
    "Stop",
    "Extend",
    "Interrupt",
    "Status",
    "SendReport",
    "ToggleMini",
    "Raise",
)

INTROSPECTION = "\n".join(
    [f'<interface name="{INTERFACE}">']
    + [
        f'  <method name="{verb}">\n    <arg type="s" direction="out"/>\n  </method>'
        for verb in VERBS
    ]
    + ["</interface>"]
)


@pyqtClassInfo("D-Bus Interface", INTERFACE)
@pyqtClassInfo("D-Bus Introspection", INTROSPECTION)
class TimerAdaptor(QDBusAbstractAdaptor):
    """The published interface. Every slot is a verb you can bind to a key."""

    def __init__(self, parent: TimerDBusService) -> None:
        super().__init__(parent)
        self._service = parent

    # D-Bus methods are PascalCase by convention.
    @pyqtSlot(result=str)
    def StartWork(self) -> str:  # noqa: N802
        """Start a focus interval."""
        return self._service.run("start_work")

    @pyqtSlot(result=str)
    def StartBreak(self) -> str:  # noqa: N802
        return self._service.run("start_break")

    @pyqtSlot(result=str)
    def Toggle(self) -> str:  # noqa: N802
        """Start if idle, pause if running, resume if paused — one hotkey."""
        return self._service.run("toggle")

    @pyqtSlot(result=str)
    def Pause(self) -> str:  # noqa: N802
        return self._service.run("pause")

    @pyqtSlot(result=str)
    def Resume(self) -> str:  # noqa: N802
        return self._service.run("resume")

    @pyqtSlot(result=str)
    def Complete(self) -> str:  # noqa: N802
        return self._service.run("complete")

    @pyqtSlot(result=str)
    def Stop(self) -> str:  # noqa: N802
        return self._service.run("stop")

    @pyqtSlot(result=str)
    def Extend(self) -> str:  # noqa: N802
        return self._service.run("extend")

    @pyqtSlot(result=str)
    def Interrupt(self) -> str:  # noqa: N802
        """Count a distraction without stopping the clock."""
        return self._service.run("interrupt")

    @pyqtSlot(result=str)
    def Status(self) -> str:  # noqa: N802
        """One line: what is running and how much is left."""
        return self._service.run("status")

    @pyqtSlot(result=str)
    def SendReport(self) -> str:  # noqa: N802
        return self._service.run("send_report")

    @pyqtSlot(result=str)
    def ToggleMini(self) -> str:  # noqa: N802
        """Show or hide the floating mini timer."""
        return self._service.run("toggle_mini")

    @pyqtSlot(result=str)
    def Raise(self) -> str:  # noqa: N802
        """Show the main window (used by a second instance)."""
        return self._service.run("raise_window")


def register(service: TimerDBusService) -> bool:
    """Own the name and export the adaptor. False means another instance has it."""
    bus = QDBusConnection.sessionBus()
    if not bus.registerService(SERVICE_NAME):
        return False
    bus.registerObject(OBJECT_PATH, service, QDBusConnection.RegisterOption.ExportAdaptors)
    log.info("D-Bus service %s%s is up as %s", SERVICE_NAME, OBJECT_PATH, INTERFACE)
    return True


def call_existing(method: str = "Raise") -> bool:
    """Poke the instance that already owns the name."""
    from PyQt6.QtDBus import QDBusInterface

    interface = QDBusInterface(SERVICE_NAME, OBJECT_PATH, INTERFACE, QDBusConnection.sessionBus())
    if not interface.isValid():
        return False
    interface.call(method)
    return True