371 lines
16 KiB
Python
371 lines
16 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.
|
|
|
|
Also owns two custom interfaces colocated at the same object path (mirrors
|
|
dunst parking org.dunstproject.cmd0 alongside its own freedesktop interface):
|
|
|
|
eu.abdelbaki.mnotifd.History1 disk-persisted notification history —
|
|
List/Get/Pop/PopLatest/Remove/Clear/
|
|
InvokeAction + HistoryAdded/Removed/Cleared
|
|
signals. See history.py for the storage side.
|
|
|
|
eu.abdelbaki.mnotifd.Controls1 push-only: ControlChanged(id, control_id,
|
|
value) for the rich embedded controls apps
|
|
can opt into via the x-mnotifd-controls hint
|
|
(see controls.py).
|
|
|
|
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 time
|
|
|
|
import gi
|
|
|
|
gi.require_version("Gtk", "4.0")
|
|
from gi.repository import Gio, GLib # noqa: E402
|
|
|
|
import config
|
|
from controls import parse_controls
|
|
from history import HistoryStore
|
|
from paths import CONTROLS_IFACE, FDN_IFACE, FDN_PATH, HISTORY_IFACE
|
|
|
|
# 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>
|
|
"""
|
|
|
|
# a{sv} per entry (not a fixed tuple shape) so new fields can be added later
|
|
# without breaking existing clients — the same shape dunst's own
|
|
# NotificationListHistory uses.
|
|
_HISTORY_INTROSPECTION_XML = """
|
|
<node>
|
|
<interface name="eu.abdelbaki.mnotifd.History1">
|
|
<method name="List">
|
|
<arg type="aa{sv}" name="entries" direction="out"/>
|
|
</method>
|
|
<method name="Get">
|
|
<arg type="u" name="id" direction="in"/>
|
|
<arg type="a{sv}" name="entry" direction="out"/>
|
|
</method>
|
|
<method name="Pop">
|
|
<arg type="u" name="id" direction="in"/>
|
|
<arg type="u" name="new_id" direction="out"/>
|
|
</method>
|
|
<method name="PopLatest">
|
|
<arg type="u" name="new_id" direction="out"/>
|
|
</method>
|
|
<method name="Remove">
|
|
<arg type="u" name="id" direction="in"/>
|
|
</method>
|
|
<method name="Clear">
|
|
</method>
|
|
<method name="InvokeAction">
|
|
<arg type="u" name="id" direction="in"/>
|
|
<arg type="s" name="action_key" direction="in"/>
|
|
</method>
|
|
<signal name="HistoryAdded">
|
|
<arg type="a{sv}" name="entry"/>
|
|
</signal>
|
|
<signal name="HistoryRemoved">
|
|
<arg type="u" name="id"/>
|
|
</signal>
|
|
<signal name="HistoryCleared">
|
|
<arg type="u" name="count"/>
|
|
</signal>
|
|
</interface>
|
|
</node>
|
|
"""
|
|
|
|
_CONTROLS_INTROSPECTION_XML = """
|
|
<node>
|
|
<interface name="eu.abdelbaki.mnotifd.Controls1">
|
|
<signal name="ControlChanged">
|
|
<arg type="u" name="id"/>
|
|
<arg type="s" name="control_id"/>
|
|
<arg type="v" name="value"/>
|
|
</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] = {}
|
|
# metadata for still-visible notifications, keyed by id — snapshotted at
|
|
# Notify() time (before render) so emit_closed() can hand a full record
|
|
# to history without window.py/notification.py needing to know about it.
|
|
self._live_meta: dict[int, dict] = {}
|
|
|
|
self._history_enabled = config.history_enabled()
|
|
self._history = HistoryStore(maxlen=config.history_length(),
|
|
persist=config.history_persist())
|
|
|
|
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)
|
|
|
|
hnode = Gio.DBusNodeInfo.new_for_xml(_HISTORY_INTROSPECTION_XML)
|
|
connection.register_object(
|
|
FDN_PATH, hnode.interfaces[0], self._on_history_method_call, None, None)
|
|
|
|
cnode = Gio.DBusNodeInfo.new_for_xml(_CONTROLS_INTROSPECTION_XML)
|
|
connection.register_object(
|
|
FDN_PATH, cnode.interfaces[0], self._on_controls_method_call, None, None)
|
|
|
|
# -- D-Bus dispatch: org.freedesktop.Notifications -------------------------
|
|
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",
|
|
"x-mnotifd-controls"],)))
|
|
elif method == "GetServerInformation":
|
|
invocation.return_value(GLib.Variant("(ssss)", (
|
|
"mnotifd", "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 ""
|
|
controls_raw = hints.get("x-mnotifd-controls")
|
|
controls = parse_controls(controls_raw)
|
|
|
|
self._live_meta[nid] = {
|
|
"id": nid, "app_name": app_name or "", "summary": summary,
|
|
"body": body, "urgency": urgency, "icon": icon,
|
|
"actions": list(actions), "controls": controls_raw,
|
|
"timestamp": time.time(),
|
|
}
|
|
|
|
self._window.show_notification(
|
|
nid, summary, body, urgency, icon, image_data, list(actions),
|
|
controls, self._emit_action, self._emit_control)
|
|
|
|
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 / on_control 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)))
|
|
meta = self._live_meta.pop(nid, None)
|
|
if meta is not None and self._history_enabled:
|
|
entry = {**meta, "reason": reason}
|
|
evicted = self._history.record(entry)
|
|
self._emit_history_signal("HistoryAdded", GLib.Variant(
|
|
"(a{sv})", (_entry_dict(entry),)))
|
|
if evicted is not None:
|
|
self._emit_history_signal("HistoryRemoved", GLib.Variant(
|
|
"(u)", (int(evicted["id"]),)))
|
|
|
|
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)
|
|
|
|
def _emit_control(self, nid: int, control_id: str, value) -> None:
|
|
self._conn.emit_signal(None, FDN_PATH, CONTROLS_IFACE, "ControlChanged",
|
|
GLib.Variant("(usv)", (nid, control_id, _variant_for(value))))
|
|
|
|
def _emit_history_signal(self, name: str, payload: GLib.Variant) -> None:
|
|
self._conn.emit_signal(None, FDN_PATH, HISTORY_IFACE, name, payload)
|
|
|
|
# -- D-Bus dispatch: eu.abdelbaki.mnotifd.History1 --------------------------
|
|
def _on_history_method_call(self, _conn, _sender, _path, _iface, method, params, invocation):
|
|
if method == "List":
|
|
entries = [_entry_dict(e) for e in self._history.list_all()]
|
|
invocation.return_value(GLib.Variant("(aa{sv})", (entries,)))
|
|
elif method == "Get":
|
|
(nid,) = params.unpack()
|
|
entry = self._history.get(int(nid))
|
|
if entry is None:
|
|
invocation.return_error_literal(
|
|
Gio.dbus_error_quark(), Gio.DBusError.FAILED,
|
|
f"no history entry with id {nid}")
|
|
else:
|
|
invocation.return_value(GLib.Variant("(a{sv})", (_entry_dict(entry),)))
|
|
elif method == "Pop":
|
|
(nid,) = params.unpack()
|
|
entry = self._history.pop(int(nid))
|
|
if entry is None:
|
|
invocation.return_error_literal(
|
|
Gio.dbus_error_quark(), Gio.DBusError.FAILED,
|
|
f"no history entry with id {nid}")
|
|
else:
|
|
self._emit_history_signal("HistoryRemoved", GLib.Variant("(u)", (int(nid),)))
|
|
invocation.return_value(GLib.Variant("(u)", (self._redisplay(entry),)))
|
|
elif method == "PopLatest":
|
|
entry = self._history.pop_latest()
|
|
if entry is None:
|
|
invocation.return_error_literal(
|
|
Gio.dbus_error_quark(), Gio.DBusError.FAILED, "history is empty")
|
|
else:
|
|
self._emit_history_signal(
|
|
"HistoryRemoved", GLib.Variant("(u)", (int(entry["id"]),)))
|
|
invocation.return_value(GLib.Variant("(u)", (self._redisplay(entry),)))
|
|
elif method == "Remove":
|
|
(nid,) = params.unpack()
|
|
if self._history.remove(int(nid)):
|
|
self._emit_history_signal("HistoryRemoved", GLib.Variant("(u)", (int(nid),)))
|
|
invocation.return_value(None)
|
|
elif method == "Clear":
|
|
removed = self._history.clear()
|
|
self._emit_history_signal("HistoryCleared", GLib.Variant("(u)", (len(removed),)))
|
|
invocation.return_value(None)
|
|
elif method == "InvokeAction":
|
|
(nid, key) = params.unpack()
|
|
entry = self._history.get(int(nid))
|
|
if entry is None:
|
|
invocation.return_error_literal(
|
|
Gio.dbus_error_quark(), Gio.DBusError.FAILED,
|
|
f"no history entry with id {nid}")
|
|
else:
|
|
# Re-emit ActionInvoked for a *historical* entry without bringing the
|
|
# card back on-screen — mnotifd (unlike dunst) doesn't void actions once
|
|
# a notification leaves the queue, since invoking one has always just
|
|
# meant "emit the signal," independent of whether a card widget exists.
|
|
self._conn.emit_signal(None, FDN_PATH, FDN_IFACE, "ActionInvoked",
|
|
GLib.Variant("(us)", (int(nid), key)))
|
|
invocation.return_value(None)
|
|
else:
|
|
invocation.return_error_literal(
|
|
Gio.dbus_error_quark(), Gio.DBusError.UNKNOWN_METHOD, method)
|
|
|
|
def _redisplay(self, entry: dict) -> int:
|
|
"""Re-render a history entry as a live card (History1.Pop/PopLatest)."""
|
|
new_nid = self._next_id
|
|
self._next_id += 1
|
|
controls = parse_controls(entry.get("controls"))
|
|
self._window.show_notification(
|
|
new_nid, entry.get("summary", ""), entry.get("body", ""),
|
|
int(entry.get("urgency", 1)), entry.get("icon", ""), None,
|
|
list(entry.get("actions") or []), controls,
|
|
self._emit_action, self._emit_control)
|
|
self._live_meta[new_nid] = {
|
|
"id": new_nid, "app_name": entry.get("app_name", ""),
|
|
"summary": entry.get("summary", ""), "body": entry.get("body", ""),
|
|
"urgency": int(entry.get("urgency", 1)), "icon": entry.get("icon", ""),
|
|
"actions": list(entry.get("actions") or []),
|
|
"controls": entry.get("controls"), "timestamp": time.time(),
|
|
}
|
|
self._arm_timer(new_nid, -1, int(entry.get("urgency", 1)))
|
|
return new_nid
|
|
|
|
# -- D-Bus dispatch: eu.abdelbaki.mnotifd.Controls1 (signal-only, no methods)
|
|
def _on_controls_method_call(self, _conn, _sender, _path, _iface, method, _params, invocation):
|
|
invocation.return_error_literal(
|
|
Gio.dbus_error_quark(), Gio.DBusError.UNKNOWN_METHOD, method)
|
|
|
|
|
|
def _entry_dict(entry: dict) -> dict:
|
|
"""A plain dict with GLib.Variant leaf values, ready to embed in an a{sv}."""
|
|
return {
|
|
"id": GLib.Variant("u", int(entry["id"])),
|
|
"app_name": GLib.Variant("s", entry.get("app_name") or ""),
|
|
"summary": GLib.Variant("s", entry.get("summary") or ""),
|
|
"body": GLib.Variant("s", entry.get("body") or ""),
|
|
"icon": GLib.Variant("s", entry.get("icon") or ""),
|
|
"urgency": GLib.Variant("i", int(entry.get("urgency", 1))),
|
|
"timestamp": GLib.Variant("d", float(entry.get("timestamp", 0.0))),
|
|
"reason": GLib.Variant("i", int(entry.get("reason", 0))),
|
|
"actions": GLib.Variant("as", list(entry.get("actions") or [])),
|
|
"controls": GLib.Variant("s", entry.get("controls") or ""),
|
|
}
|
|
|
|
|
|
def _variant_for(value) -> GLib.Variant:
|
|
if isinstance(value, bool):
|
|
return GLib.Variant("b", value)
|
|
if isinstance(value, (int, float)):
|
|
return GLib.Variant("d", float(value))
|
|
return GLib.Variant("s", str(value))
|