300 lines
11 KiB
Python
300 lines
11 KiB
Python
|
# -*- coding: utf-8 -*-
|
||
|
|
"""
|
||
|
|
Notification backend for Linux
|
||
|
|
|
||
|
|
Includes an implementation to send desktop notifications over Dbus. Responding to user
|
||
|
|
interaction with a notification requires a running asyncio event loop.
|
||
|
|
"""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import logging
|
||
|
|
from typing import TypeVar
|
||
|
|
|
||
|
|
from bidict import bidict
|
||
|
|
from dbus_fast.aio.message_bus import MessageBus
|
||
|
|
from dbus_fast.aio.proxy_object import ProxyInterface
|
||
|
|
from dbus_fast.errors import DBusError
|
||
|
|
from dbus_fast.signature import Variant
|
||
|
|
|
||
|
|
from ..common import Capability, Notification, Urgency
|
||
|
|
from .base import DesktopNotifierBackend
|
||
|
|
|
||
|
|
__all__ = ["DBusDesktopNotifier"]
|
||
|
|
|
||
|
|
logger = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
T = TypeVar("T")
|
||
|
|
|
||
|
|
NOTIFICATION_CLOSED_EXPIRED = 1
|
||
|
|
NOTIFICATION_CLOSED_DISMISSED = 2
|
||
|
|
NOTIFICATION_CLOSED_PROGRAMMATICALLY = 3
|
||
|
|
NOTIFICATION_CLOSED_UNDEFINED = 4
|
||
|
|
|
||
|
|
|
||
|
|
class DBusDesktopNotifier(DesktopNotifierBackend):
|
||
|
|
"""DBus notification backend for Linux
|
||
|
|
|
||
|
|
This implements the org.freedesktop.Notifications standard. The DBUS connection is
|
||
|
|
created in a thread with a running asyncio loop to handle clicked notifications.
|
||
|
|
|
||
|
|
:param app_name: The name of the app.
|
||
|
|
"""
|
||
|
|
|
||
|
|
to_native_urgency = {
|
||
|
|
Urgency.Low: Variant("y", 0),
|
||
|
|
Urgency.Normal: Variant("y", 1),
|
||
|
|
Urgency.Critical: Variant("y", 2),
|
||
|
|
}
|
||
|
|
|
||
|
|
supported_hint_signatures = {"a{sv}", "a{ss}"}
|
||
|
|
|
||
|
|
def __init__(self, app_name: str) -> None:
|
||
|
|
super().__init__(app_name)
|
||
|
|
self.interface: ProxyInterface | None = None
|
||
|
|
self._platform_to_interface_notification_identifier: bidict[int, str] = bidict()
|
||
|
|
|
||
|
|
async def request_authorisation(self) -> bool:
|
||
|
|
"""
|
||
|
|
Request authorisation to send notifications.
|
||
|
|
|
||
|
|
:returns: Whether authorisation has been granted.
|
||
|
|
"""
|
||
|
|
return True
|
||
|
|
|
||
|
|
async def has_authorisation(self) -> bool:
|
||
|
|
"""
|
||
|
|
Whether we have authorisation to send notifications.
|
||
|
|
"""
|
||
|
|
return True
|
||
|
|
|
||
|
|
async def _init_dbus(self) -> ProxyInterface:
|
||
|
|
self.bus = await MessageBus().connect()
|
||
|
|
introspection = await self.bus.introspect(
|
||
|
|
"org.freedesktop.Notifications", "/org/freedesktop/Notifications"
|
||
|
|
)
|
||
|
|
self.proxy_object = self.bus.get_proxy_object(
|
||
|
|
"org.freedesktop.Notifications",
|
||
|
|
"/org/freedesktop/Notifications",
|
||
|
|
introspection,
|
||
|
|
)
|
||
|
|
self.interface = self.proxy_object.get_interface(
|
||
|
|
"org.freedesktop.Notifications"
|
||
|
|
)
|
||
|
|
|
||
|
|
# Some older interfaces may not support notification actions.
|
||
|
|
if hasattr(self.interface, "on_notification_closed"):
|
||
|
|
self.interface.on_notification_closed(self._on_closed)
|
||
|
|
|
||
|
|
if hasattr(self.interface, "on_action_invoked"):
|
||
|
|
self.interface.on_action_invoked(self._on_action)
|
||
|
|
|
||
|
|
return self.interface
|
||
|
|
|
||
|
|
async def _send(self, notification: Notification) -> None:
|
||
|
|
"""
|
||
|
|
Asynchronously sends a notification via the Dbus interface.
|
||
|
|
|
||
|
|
:param notification: Notification to send.
|
||
|
|
"""
|
||
|
|
if not self.interface:
|
||
|
|
self.interface = await self._init_dbus()
|
||
|
|
|
||
|
|
# The "default" action is typically invoked when clicking on the
|
||
|
|
# notification body itself, see
|
||
|
|
# https://specifications.freedesktop.org/notification-spec. There are some
|
||
|
|
# exceptions though, such as XFCE, where this will result in a separate
|
||
|
|
# button. If no label name is provided in XFCE, it will result in a default
|
||
|
|
# symbol being used. We therefore don't specify a label name.
|
||
|
|
actions = ["default", ""]
|
||
|
|
|
||
|
|
for button in notification.buttons:
|
||
|
|
actions += [button.identifier, button.title]
|
||
|
|
|
||
|
|
hints_v: dict[str, Variant] = dict()
|
||
|
|
hints_v["urgency"] = self.to_native_urgency[notification.urgency]
|
||
|
|
|
||
|
|
if notification.sound:
|
||
|
|
if notification.sound.is_named():
|
||
|
|
hints_v["sound-name"] = Variant("s", "message-new-instant")
|
||
|
|
else:
|
||
|
|
hints_v["sound-file"] = Variant("s", notification.sound.as_uri())
|
||
|
|
|
||
|
|
if notification.attachment:
|
||
|
|
hints_v["image-path"] = Variant("s", notification.attachment.as_uri())
|
||
|
|
|
||
|
|
# The current notification spec defines hints as a Dbus dictionary type 'a{sv}',
|
||
|
|
# represented in Python as dict[str, Variant]. However, some older notification
|
||
|
|
# servers expect 'a{ss}' (Python dict[str, str]). We therefore check the
|
||
|
|
# expected argument type at runtime and cast arguments accordingly.
|
||
|
|
# See https://github.com/samschott/desktop-notifier/issues/143.
|
||
|
|
hints_signature = get_hints_signature(self.interface)
|
||
|
|
|
||
|
|
if hints_signature == "":
|
||
|
|
logger.warning("Notification server not supported")
|
||
|
|
return
|
||
|
|
|
||
|
|
hints: dict[str, str] | dict[str, Variant]
|
||
|
|
|
||
|
|
if hints_signature == "a{sv}":
|
||
|
|
hints = hints_v
|
||
|
|
elif hints_signature == "a{ss}":
|
||
|
|
hints = {k: str(v.value) for k, v in hints_v.items()}
|
||
|
|
else:
|
||
|
|
hints = {}
|
||
|
|
|
||
|
|
timeout = notification.timeout * 1000 if notification.timeout != -1 else -1
|
||
|
|
if notification.icon:
|
||
|
|
if notification.icon.is_named():
|
||
|
|
icon = notification.icon.name
|
||
|
|
else:
|
||
|
|
icon = notification.icon.as_uri()
|
||
|
|
else:
|
||
|
|
icon = ""
|
||
|
|
|
||
|
|
# dbus_next proxy APIs are generated at runtime. Silence the type checker but
|
||
|
|
# raise an AttributeError if required.
|
||
|
|
platform_id = await self.interface.call_notify( # type:ignore[attr-defined]
|
||
|
|
self.app_name,
|
||
|
|
0,
|
||
|
|
icon,
|
||
|
|
notification.title,
|
||
|
|
notification.message,
|
||
|
|
actions,
|
||
|
|
hints,
|
||
|
|
timeout,
|
||
|
|
)
|
||
|
|
self._platform_to_interface_notification_identifier[platform_id] = (
|
||
|
|
notification.identifier
|
||
|
|
)
|
||
|
|
|
||
|
|
async def _clear(self, identifier: str) -> None:
|
||
|
|
"""
|
||
|
|
Asynchronously removes a notification from the notification center
|
||
|
|
"""
|
||
|
|
if not self.interface:
|
||
|
|
return
|
||
|
|
|
||
|
|
platform_id = self._platform_to_interface_notification_identifier.inverse[
|
||
|
|
identifier
|
||
|
|
]
|
||
|
|
|
||
|
|
try:
|
||
|
|
# dbus_next proxy APIs are generated at runtime. Silence the type checker
|
||
|
|
# but raise an AttributeError if required.
|
||
|
|
await self.interface.call_close_notification( # type:ignore[attr-defined]
|
||
|
|
platform_id
|
||
|
|
)
|
||
|
|
except DBusError:
|
||
|
|
# Notification was already closed.
|
||
|
|
# See https://specifications.freedesktop.org/notification-spec/latest/protocol.html#command-close-notification
|
||
|
|
pass
|
||
|
|
|
||
|
|
try:
|
||
|
|
del self._platform_to_interface_notification_identifier.inverse[identifier]
|
||
|
|
except KeyError:
|
||
|
|
# Popping may have been handled already by _on_close callback.
|
||
|
|
pass
|
||
|
|
|
||
|
|
async def _clear_all(self) -> None:
|
||
|
|
"""
|
||
|
|
Asynchronously clears all notifications from notification center
|
||
|
|
"""
|
||
|
|
if not self.interface:
|
||
|
|
return
|
||
|
|
|
||
|
|
for identifier in list(
|
||
|
|
self._platform_to_interface_notification_identifier.values()
|
||
|
|
):
|
||
|
|
await self._clear(identifier)
|
||
|
|
|
||
|
|
# Note that _on_action and _on_closed might be called for the same notification
|
||
|
|
# with some notification servers. This is not a problem because the _on_action
|
||
|
|
# call will come first, in which case we are no longer interested in calling the
|
||
|
|
# _on_closed callback.
|
||
|
|
|
||
|
|
def _on_action(self, nid: int, action_key: str) -> None:
|
||
|
|
"""
|
||
|
|
Called when the user performs a notification action. This will invoke the
|
||
|
|
handler callback.
|
||
|
|
|
||
|
|
:param nid: The platform's notification ID as an integer.
|
||
|
|
:param action_key: A string identifying the action to take. We choose those keys
|
||
|
|
ourselves when scheduling the notification.
|
||
|
|
"""
|
||
|
|
identifier = self._platform_to_interface_notification_identifier.pop(nid, "")
|
||
|
|
notification = self._clear_notification_from_cache(identifier)
|
||
|
|
|
||
|
|
if not notification:
|
||
|
|
return
|
||
|
|
|
||
|
|
if action_key == "default":
|
||
|
|
self.handle_clicked(identifier, notification)
|
||
|
|
return
|
||
|
|
|
||
|
|
self.handle_button(identifier, action_key, notification)
|
||
|
|
|
||
|
|
def _on_closed(self, nid: int, reason: int) -> None:
|
||
|
|
"""
|
||
|
|
Called when the user closes a notification. This will invoke the registered
|
||
|
|
callback.
|
||
|
|
|
||
|
|
:param nid: The platform's notification ID as an integer.
|
||
|
|
:param reason: An integer describing the reason why the notification was closed.
|
||
|
|
"""
|
||
|
|
identifier = self._platform_to_interface_notification_identifier.pop(nid, "")
|
||
|
|
notification = self._clear_notification_from_cache(identifier)
|
||
|
|
|
||
|
|
if not notification:
|
||
|
|
return
|
||
|
|
|
||
|
|
if reason == NOTIFICATION_CLOSED_DISMISSED:
|
||
|
|
self.handle_dismissed(identifier, notification)
|
||
|
|
|
||
|
|
async def get_capabilities(self) -> frozenset[Capability]:
|
||
|
|
if not self.interface:
|
||
|
|
self.interface = await self._init_dbus()
|
||
|
|
|
||
|
|
capabilities = {
|
||
|
|
Capability.APP_NAME,
|
||
|
|
Capability.ICON,
|
||
|
|
Capability.TITLE,
|
||
|
|
Capability.TIMEOUT,
|
||
|
|
Capability.URGENCY,
|
||
|
|
}
|
||
|
|
|
||
|
|
# Capabilities supported by some notification servers.
|
||
|
|
# See https://specifications.freedesktop.org/notification-spec/notification-spec-latest.html#protocol.
|
||
|
|
if hasattr(self.interface, "on_notification_closed"):
|
||
|
|
capabilities.add(Capability.ON_CLICKED)
|
||
|
|
capabilities.add(Capability.ON_DISMISSED)
|
||
|
|
|
||
|
|
cps = await self.interface.call_get_capabilities() # type:ignore[attr-defined]
|
||
|
|
|
||
|
|
if "actions" in cps:
|
||
|
|
capabilities.add(Capability.BUTTONS)
|
||
|
|
if "body" in cps:
|
||
|
|
capabilities.add(Capability.MESSAGE)
|
||
|
|
if "sound" in cps:
|
||
|
|
capabilities.add(Capability.SOUND)
|
||
|
|
capabilities.add(Capability.SOUND_NAME)
|
||
|
|
|
||
|
|
hints_signature = get_hints_signature(self.interface)
|
||
|
|
if hints_signature not in self.supported_hint_signatures:
|
||
|
|
# Any hint-based capabilities are not supported because we got an unexpected
|
||
|
|
# DBus interface.
|
||
|
|
capabilities.discard(Capability.SOUND)
|
||
|
|
capabilities.discard(Capability.SOUND_NAME)
|
||
|
|
capabilities.discard(Capability.URGENCY)
|
||
|
|
|
||
|
|
return frozenset(capabilities)
|
||
|
|
|
||
|
|
|
||
|
|
def get_hints_signature(interface: ProxyInterface) -> str:
|
||
|
|
"""Returns the dbus type signature for the hints argument"""
|
||
|
|
methods = interface.introspection.methods
|
||
|
|
notify_method = next(m for m in methods if m.name == "Notify")
|
||
|
|
try:
|
||
|
|
return str(notify_method.in_args[6].signature)
|
||
|
|
except IndexError:
|
||
|
|
return ""
|