147 lines
5.9 KiB
Python
147 lines
5.9 KiB
Python
"""org.freedesktop.Notifications D-Bus service — the well-known name dunst used
|
|
to own. Parses each Notify call and hands it to the stack window (window.py) to
|
|
render as a holographic card; owns per-notification expiry timers and emits the
|
|
spec's NotificationClosed / ActionInvoked signals.
|
|
|
|
Deliberately small: the visible/interactive spec subset (body markup, actions,
|
|
icons, urgency, replaces_id, timeouts, persistence) — no sound/markup-hint
|
|
gymnastics.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import gi
|
|
|
|
gi.require_version("Gtk", "4.0")
|
|
from gi.repository import Gio, GLib # noqa: E402
|
|
|
|
from paths import FDN_IFACE, FDN_PATH
|
|
|
|
# reasons per the spec: 1 expired, 2 dismissed by user, 3 closed by call, 4 undefined
|
|
_INTROSPECTION_XML = """
|
|
<node>
|
|
<interface name="org.freedesktop.Notifications">
|
|
<method name="GetCapabilities">
|
|
<arg type="as" name="capabilities" direction="out"/>
|
|
</method>
|
|
<method name="Notify">
|
|
<arg type="s" name="app_name" direction="in"/>
|
|
<arg type="u" name="replaces_id" direction="in"/>
|
|
<arg type="s" name="app_icon" direction="in"/>
|
|
<arg type="s" name="summary" direction="in"/>
|
|
<arg type="s" name="body" direction="in"/>
|
|
<arg type="as" name="actions" direction="in"/>
|
|
<arg type="a{sv}" name="hints" direction="in"/>
|
|
<arg type="i" name="expire_timeout" direction="in"/>
|
|
<arg type="u" name="id" direction="out"/>
|
|
</method>
|
|
<method name="CloseNotification">
|
|
<arg type="u" name="id" direction="in"/>
|
|
</method>
|
|
<method name="GetServerInformation">
|
|
<arg type="s" name="name" direction="out"/>
|
|
<arg type="s" name="vendor" direction="out"/>
|
|
<arg type="s" name="version" direction="out"/>
|
|
<arg type="s" name="spec_version" direction="out"/>
|
|
</method>
|
|
<signal name="NotificationClosed">
|
|
<arg type="u" name="id"/>
|
|
<arg type="u" name="reason"/>
|
|
</signal>
|
|
<signal name="ActionInvoked">
|
|
<arg type="u" name="id"/>
|
|
<arg type="s" name="action_key"/>
|
|
</signal>
|
|
</interface>
|
|
</node>
|
|
"""
|
|
|
|
# server default timeouts (ms) by urgency when the client passes -1
|
|
_DEFAULT_TIMEOUT = {0: 5000, 1: 8000, 2: 0} # low / normal / critical(never)
|
|
|
|
|
|
class NotificationServer:
|
|
def __init__(self, connection: Gio.DBusConnection, window) -> None:
|
|
self._conn = connection
|
|
self._window = window
|
|
self._next_id = 1
|
|
self._timers: dict[int, int] = {}
|
|
|
|
node = Gio.DBusNodeInfo.new_for_xml(_INTROSPECTION_XML)
|
|
self._iface = node.interfaces[0]
|
|
connection.register_object(
|
|
FDN_PATH, self._iface, self._on_method_call, None, None)
|
|
|
|
# -- D-Bus dispatch -------------------------------------------------------
|
|
def _on_method_call(self, _conn, _sender, _path, _iface, method, params, invocation):
|
|
if method == "Notify":
|
|
invocation.return_value(GLib.Variant("(u)", (self._notify(params),)))
|
|
elif method == "CloseNotification":
|
|
(nid,) = params.unpack()
|
|
self._window.close_notification(int(nid), reason=3)
|
|
self._cancel_timer(int(nid))
|
|
invocation.return_value(None)
|
|
elif method == "GetCapabilities":
|
|
invocation.return_value(GLib.Variant("(as)", (
|
|
["body", "body-markup", "icon-static", "actions", "persistence"],)))
|
|
elif method == "GetServerInformation":
|
|
invocation.return_value(GLib.Variant("(ssss)", (
|
|
"beacon", "abdelbaki.eu", "1.0", "1.2")))
|
|
else:
|
|
invocation.return_error_literal(
|
|
Gio.dbus_error_quark(), Gio.DBusError.UNKNOWN_METHOD, method)
|
|
|
|
def _notify(self, params) -> int:
|
|
(app_name, replaces_id, app_icon, summary, body,
|
|
actions, hints, expire_timeout) = params.unpack()
|
|
|
|
nid = int(replaces_id) if replaces_id else self._next_id
|
|
if not replaces_id:
|
|
self._next_id += 1
|
|
|
|
urgency = int(hints.get("urgency", 1))
|
|
image_data = (hints.get("image-data") or hints.get("image_data")
|
|
or hints.get("icon_data"))
|
|
image_path = hints.get("image-path") or hints.get("image_path")
|
|
icon = image_path or app_icon or ""
|
|
|
|
self._window.show_notification(
|
|
nid, summary, body, urgency, icon, image_data, list(actions),
|
|
self._emit_action)
|
|
|
|
self._arm_timer(nid, int(expire_timeout), urgency)
|
|
return nid
|
|
|
|
# -- expiry timers --------------------------------------------------------
|
|
def _arm_timer(self, nid: int, expire_timeout: int, urgency: int) -> None:
|
|
self._cancel_timer(nid)
|
|
if expire_timeout < 0:
|
|
ms = _DEFAULT_TIMEOUT.get(urgency, 8000)
|
|
else:
|
|
ms = expire_timeout
|
|
if ms <= 0:
|
|
return # 0 = never expire (or critical default)
|
|
self._timers[nid] = GLib.timeout_add(ms, self._on_expire, nid)
|
|
|
|
def _on_expire(self, nid: int) -> bool:
|
|
self._timers.pop(nid, None)
|
|
self._window.close_notification(nid, reason=1) # 1 = expired
|
|
return False
|
|
|
|
def _cancel_timer(self, nid: int) -> None:
|
|
tid = self._timers.pop(nid, None)
|
|
if tid is not None:
|
|
GLib.source_remove(tid)
|
|
|
|
# -- signals (also the window's on_closed / on_action callbacks) -----------
|
|
def emit_closed(self, nid: int, reason: int) -> None:
|
|
self._cancel_timer(nid)
|
|
self._conn.emit_signal(None, FDN_PATH, FDN_IFACE, "NotificationClosed",
|
|
GLib.Variant("(uu)", (nid, reason)))
|
|
|
|
def _emit_action(self, nid: int, key: str) -> None:
|
|
self._conn.emit_signal(None, FDN_PATH, FDN_IFACE, "ActionInvoked",
|
|
GLib.Variant("(us)", (nid, key)))
|
|
# dismiss the card once its action fired, matching typical daemon behaviour
|
|
self._window.close_notification(nid, reason=2)
|