Replace python-dbus with jeepney for notifications

This commit is contained in:
Kovid Goyal 2021-06-24 16:40:31 +05:30
parent 9fd0cf0cc7
commit 58d392ca65
No known key found for this signature in database
GPG Key ID: 06BC317B515ACE7C

View File

@ -7,15 +7,15 @@ __copyright__ = '2009, Kovid Goyal <kovid@kovidgoyal.net>'
__docformat__ = 'restructuredtext en'
import time
import sys
from contextlib import suppress
from functools import lru_cache
from calibre import prints
from calibre.constants import DEBUG, get_osx_version, islinux, ismacos
from calibre.constants import DEBUG, __appname__, get_osx_version, islinux, ismacos
from polyglot.builtins import unicode_type
class Notifier(object):
class Notifier:
DEFAULT_TIMEOUT = 5000
@ -37,47 +37,76 @@ def icon(data=False):
class DBUSNotifier(Notifier):
def __init__(self, session_bus):
self.ok, self.err = True, None
server, path, interface = self.SERVICE
if DEBUG:
start = time.time()
prints('Looking for desktop notifier support from:', server)
def __init__(self):
self.initialized = False
def initialize(self):
from jeepney.io.blocking import open_dbus_connection
if self.initialized:
return
self.initialized = True
self.ok = False
try:
import dbus
self.dbus = dbus
self._notify = dbus.Interface(session_bus.get_object(server, path), interface)
except Exception as err:
self.ok = False
self.err = unicode_type(err)
self.connection = open_dbus_connection(bus='SESSION')
except Exception:
return
with suppress(Exception):
self.ok = self.initialize_fdo()
if self.ok:
self.notify = self.fdo_notify
return
if DEBUG:
prints(server, 'found' if self.ok else 'not found', 'in', '%.1f' % (time.time() - start), 'seconds')
print('Failed to connect to FDO Notifications service', file=sys.stderr)
with suppress(Exception):
self.ok = self.initialize_portal()
if self.ok:
self.notify = self.portal_notify
else:
print('Failed to connect to Portal Notifications service', file=sys.stderr)
def initialize_fdo(self):
from jeepney import DBusAddress, MessageType, new_method_call
self.address = DBusAddress(
'/org/freedesktop/Notifications',
bus_name='org.freedesktop.Notifications',
interface='org.freedesktop.Notifications')
class FDONotifier(DBUSNotifier):
msg = new_method_call(self.address, 'GetCapabilities')
reply = self.connection.send_and_get_reply(msg)
return bool(reply and reply.header.message_type is MessageType.method_return)
SERVICE = 'org.freedesktop.Notifications', '/org/freedesktop/Notifications', 'org.freedesktop.Notifications'
def initialize_portal(self):
from jeepney import DBusAddress, MessageType, Properties
self.address = DBusAddress(
'/org/freedesktop/portal/desktop',
bus_name='org.freedesktop.portal.Desktop',
interface='org.freedesktop.portal.Notification')
p = Properties(self.address)
msg = p.get('version')
reply = self.connection.send_and_get_reply(msg)
return bool(reply and reply.header.message_type is MessageType.method_return and reply.body[0][1] >= 1)
def __call__(self, body, summary=None, replaces_id=None, timeout=0):
if replaces_id is None:
replaces_id = self.dbus.UInt32()
def fdo_notify(self, body, summary=None, replaces_id=None, timeout=0):
from jeepney import new_method_call
timeout, body, summary = self.get_msg_parms(timeout, body, summary)
msg = new_method_call(
self.address, 'Notify', 'susssasa{sv}i',
(__appname__,
replaces_id or 0,
icon(),
summary,
body,
[], {}, # Actions, hints
timeout,
))
try:
self._notify.Notify('calibre', replaces_id, icon(), summary, body,
self.dbus.Array(signature='s'), self.dbus.Dictionary({"desktop-entry": "calibre-gui"}, signature='sv'),
timeout)
self.connection.send(msg)
except Exception:
import traceback
traceback.print_exc()
class XDPNotifier(DBUSNotifier):
SERVICE = 'org.freedesktop.portal.Desktop', '/org/freedesktop/portal/desktop', 'org.freedesktop.portal.Notification'
def __call__(self, body, summary=None, replaces_id=None, timeout=0):
if replaces_id is None:
replaces_id = self.dbus.UInt32()
def portal_notify(self, body, summary=None, replaces_id=None, timeout=0):
from jeepney import new_method_call
_, body, summary = self.get_msg_parms(timeout, body, summary)
# Note: This backend does not natively support the notion of timeouts
#
@ -93,27 +122,33 @@ class XDPNotifier(DBUSNotifier):
# Doing that however, requires Calibre to first be converted to use
# its AppID everywhere and then we still need a fallback for portable
# installations.
msg = new_method_call(
self.address, 'AddNotification', 'sa{sv}', (
str(replaces_id or 0),
{
"title": ('s', summary),
"body": ('s', body),
"icon": (
'(sv)',
(
"bytes",
('ay', icon(data=True))
)
),
}))
try:
self._notify.AddNotification(str(replaces_id), self.dbus.Dictionary({
"title": self.dbus.String(summary),
"body": self.dbus.String(body),
"icon": self.dbus.Struct(("bytes", self.dbus.ByteArray(icon(data=True), variant_level=1)), signature='sv'),
}, signature='sv'))
self.connection.send(msg)
except Exception:
import traceback
traceback.print_exc()
def get_dbus_notifier():
import dbus
session_bus = dbus.SessionBus()
names = frozenset(session_bus.list_names())
for srv in (FDONotifier, XDPNotifier):
if srv.SERVICE[0] in names:
ans = srv(session_bus)
if ans.ok:
return ans
def __call__(self, body, summary=None, replaces_id=None, timeout=0):
self.initialize()
if not self.ok:
if DEBUG:
print('Failed to connect to any notification service', file=sys.stderr)
return
self.notify(body, summary, replaces_id, timeout)
class QtNotifier(Notifier):
@ -178,12 +213,7 @@ class AppleNotifier(Notifier):
def get_notifier(systray=None):
ans = None
if islinux:
try:
ans = get_dbus_notifier()
except Exception:
import traceback
traceback.print_exc()
ans = None
ans = DBUSNotifier()
elif ismacos:
if get_osx_version() >= (10, 8, 0):
ans = AppleNotifier()
@ -200,6 +230,10 @@ def get_notifier(systray=None):
return ans
if __name__ == '__main__':
def hello():
n = get_notifier()
n('hello')
if __name__ == '__main__':
hello()