Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
121 changes: 50 additions & 71 deletions blueman/gui/Notification.py
Original file line number Diff line number Diff line change
@@ -1,57 +1,43 @@
from collections.abc import Callable, Iterable
from gettext import gettext as _
from typing import NamedTuple

import gi
gi.require_version("Gtk", "3.0")
gi.require_version("Gdk", "3.0")
from gi.repository import Gtk
from gi.repository import Gdk
from gi.repository import GdkPixbuf
from gi.repository import GLib
from gi.repository import Gio
from blueman.gui.GtkAnimation import AnimBase
import logging

OPACITY_START = 0.7


class Fade(AnimBase):
def __init__(self, window: Gtk.Window) -> None:
super().__init__(state=OPACITY_START)
self.window = window

def state_changed(self, state: float) -> None:
self.window.props.opacity = state
class NotificationAction(NamedTuple):
id: str
name: str
callback: Callable[[str], None] | None = None


class _NotificationDialog(Gtk.MessageDialog):
def __init__(self, summary: str, message: str, _timeout: int = -1, _transient: bool = False,
actions: Iterable[tuple[str, str]] | None = None,
actions_cb: Callable[[str], None] | None = None, icon_name: str | None = None,
actions: Iterable[NotificationAction] | None = None, icon_name: str | None = None,
image_data: GdkPixbuf.Pixbuf | None = None) -> None:
super().__init__(parent=None, type=Gtk.MessageType.QUESTION,
buttons=Gtk.ButtonsType.NONE, text=None)

self.set_name("NotificationDialog")
i = 100
self.actions_supported = True
self.actions: dict[int, str] = {}
self.callback = actions_cb
if actions:
for a in actions:
action_id = a[0]
action_name = a[1]
self.actions: dict[int, NotificationAction] = {}

self.actions[i] = action_id
self.add_button(action_name, i)
i += 1
if actions is None:
actions = [NotificationAction("close", _("_Close"), lambda s: self.close())]

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

self.close() should be redundant here, as do_response already destroys.


self.actions[Gtk.ResponseType.DELETE_EVENT] = "close"

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is fired by the window manager, e.g., in reaction to Esc or a request to close the window. I think those controls are broken now. Also, actions_cb fired for it (and any other event), and at least BlueAgent.py's on_auth_action relies on that.

for action in actions:
self.add_action(action)

self.props.secondary_use_markup = True
self.resize(350, 50)

self.fader = Fade(self)

self.props.skip_taskbar_hint = False

self.props.title = summary
Expand All @@ -65,47 +51,38 @@ def __init__(self, summary: str, message: str, _timeout: int = -1, _transient: b
elif image_data:
self.set_icon_from_pixbuf(image_data)

self.connect("response", self.dialog_response)
self.props.icon_name = "blueman"

self.entered = False

def on_enter(_widget: "_NotificationDialog", _event: Gdk.Event) -> bool:
if self.get_window() == Gdk.Window.at_pointer()[0] or not self.entered:
self.fader.animate(start=self.fader.get_state(), end=1.0, duration=500)
self.entered = True
return False

def on_leave(_widget: "_NotificationDialog", _event: Gdk.Event) -> bool:
if not Gdk.Window.at_pointer():
self.entered = False
self.fader.animate(start=self.fader.get_state(), end=OPACITY_START, duration=500)
return False

self.connect("enter-notify-event", on_enter)
self.connect("leave-notify-event", on_leave)

def set_message(self, message: str) -> None:
self.props.secondary_text = message

def set_notification_icon(self, icon_name: str) -> None:
self.set_icon_from_icon_name(icon_name, 48)

def dialog_response(self, _dialog: Gtk.Dialog, response: int) -> None:
if self.callback:
self.callback(self.actions[response])
self.hide()
def do_response(self, response: int) -> None:
action = self.actions.pop(response, None)
if action is None:
logging.error(f"Unhandled response {response}")
return

if action.callback is not None:
action.callback(action.id)
self.destroy()

def show(self) -> None:
self.set_opacity(OPACITY_START)
self.present()
self.set_opacity(OPACITY_START)

def close(self) -> None:
self.hide()
self.destroy()

def add_action(self, _action_id: str, _label: str, _callback: Callable[[str], None] | None = None) -> None:
logging.warning("stub")
def add_action(self, action: NotificationAction) -> None:
if not self.actions:
response_id = 100
else:
response_id = max(self.actions.keys()) + 1

self.actions[response_id] = action
self.add_button(action.name, response_id)

def set_icon_from_pixbuf(self, pixbuf: GdkPixbuf.Pixbuf) -> None:
im = Gtk.Image.new_from_pixbuf(pixbuf)
Expand All @@ -120,8 +97,7 @@ def set_icon_from_icon_name(self, icon_name: str, size: int) -> None:

class _NotificationBubble(Gio.DBusProxy):
def __init__(self, summary: str, message: str, timeout: int = -1, transient: bool = False,
actions: Iterable[tuple[str, str]] | None = None,
actions_cb: Callable[[str], None] | None = None, icon_name: str | None = None,
actions: Iterable[NotificationAction] | None = None, icon_name: str | None = None,
image_data: GdkPixbuf.Pixbuf | None = None) -> None:
super().__init__(
g_name='org.freedesktop.Notifications',
Expand All @@ -134,8 +110,8 @@ def __init__(self, summary: str, message: str, timeout: int = -1, transient: boo

self._app_name = 'blueman'
self._app_icon = ''
self._actions: list[str] = []
self._callbacks: dict[str, Callable[[str], None]] = {}
self._actions: dict[str, NotificationAction] = {}
self._notification_actions: list[str] = []
self._hints: dict[str, GLib.Variant] = {}

# hint : (variant format, spec version)
Expand Down Expand Up @@ -190,9 +166,9 @@ def __init__(self, summary: str, message: str, timeout: int = -1, transient: boo

self.set_hint(key, (w, h, stride, alpha, bits, channel, data))

if actions:
if actions is not None:
for action in actions:
self.add_action(action[0], action[1], actions_cb)
self.add_action(action)

self._capabilities = self.GetCapabilities()

Expand Down Expand Up @@ -231,14 +207,13 @@ def remove_hint(self, key: str) -> None:
def clear_hints(self) -> None:
self._hints = {}

def add_action(self, action_id: str, label: str, callback: Callable[[str], None] | None = None) -> None:
self._actions.extend([action_id, label])
if callback:
self._callbacks[action_id] = callback
def add_action(self, action: NotificationAction) -> None:
self._actions[action.id] = action
self._notification_actions.extend([action.id, action.name])

def clear_actions(self) -> None:
self._actions = []
self._callbacks = {}
self._actions.clear()
self._notification_actions.clear()

def do_g_signal(self, _sender_name: str, signal_name: str, params: GLib.Variant) -> None:
notif_id, signal_val = params.unpack()
Expand All @@ -257,13 +232,18 @@ def do_g_signal(self, _sender_name: str, signal_name: str, params: GLib.Variant)
elif signal_val == 4:
logging.debug('Undefined/reserved reasons.')
elif signal_name == 'ActionInvoked':
if signal_val in self._callbacks:
self._callbacks[signal_val](signal_val)
action = self._actions.pop(signal_val, None)
if action is None:
logging.error(f"Unhandled action {signal_val}")
return

if action.callback is not None:
action.callback(action.id)

def show(self) -> None:
replace_id = self._return_id if self._return_id else 0
return_id = self.Notify('(susssasa{sv}i)', self._app_name, replace_id, self._app_icon,
self._summary, self._body, self._actions, self._hints,
self._summary, self._body, self._notification_actions, self._hints,
self._timeout)
self._return_id = return_id

Expand All @@ -278,8 +258,7 @@ def Notification(
message: str,
timeout: int = -1,
transient: bool = False,
actions: Iterable[tuple[str, str]] | None = None,
actions_cb: Callable[[str], None] | None = None,
actions: Iterable[NotificationAction] | None = None,
icon_name: str | None = None,
image_data: GdkPixbuf.Pixbuf | None = None
) -> _NotificationBubble | _NotificationDialog:
Expand All @@ -302,4 +281,4 @@ def Notification(
else:
klass = _NotificationBubble

return klass(summary, message, timeout, transient, actions, actions_cb, icon_name, image_data)
return klass(summary, message, timeout, transient, actions, icon_name, image_data)
21 changes: 12 additions & 9 deletions blueman/main/applet/BluezAgent.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@
from blueman.bluez.Device import Device
from blueman.bluez.AgentManager import AgentManager
from blueman.Sdp import ServiceUUID
from blueman.gui.Notification import Notification, _NotificationBubble, _NotificationDialog
from blueman.gui.Notification import Notification, _NotificationBubble, _NotificationDialog, NotificationAction
from blueman.main.Builder import Builder
from blueman.main.DbusService import DbusService, DbusError

Expand Down Expand Up @@ -227,10 +227,13 @@ def on_confirm_action(action: str) -> None:

if passkey:
notify_message += "\n" + _("Confirm value for authentication:") + f" <b>{passkey:06}</b>"
actions = [("confirm", _("Confirm")), ("deny", _("Deny"))]
actions = [
NotificationAction("confirm", _("Confirm"), on_confirm_action),
NotificationAction("deny", _("Deny"), on_confirm_action)
]

self._notification = Notification("Bluetooth", notify_message, 0,
actions=actions, actions_cb=on_confirm_action, icon_name="blueman")
self._notification = Notification("Bluetooth", notify_message, 0, actions=actions,
icon_name="blueman")
self._notification.show()

def _on_request_authorization(self, object_path: ObjectPath, ok: Callable[[], None],
Expand All @@ -256,11 +259,11 @@ def on_auth_action(action: str) -> None:
service = ServiceUUID(uuid).name
notify_message = \
_("Authorization request for:") + f"\n{dev_str}\n" + _("Service:") + f" <b>{service}</b>"
actions = [("always", _("Always accept")),
("accept", _("Accept")),
("deny", _("Deny"))]
actions = [
NotificationAction("always", _("Always accept"), on_auth_action),
NotificationAction("accept", _("Accept"), on_auth_action),
NotificationAction("deny", _("Deny"), on_auth_action)]

n = Notification(_("Bluetooth Authentication"), notify_message, 0,
actions=actions, actions_cb=on_auth_action, icon_name="blueman")
n = Notification(_("Bluetooth Authentication"), notify_message, 0, actions=actions, icon_name="blueman")
n.show()
self._service_notifications.append(n)
16 changes: 11 additions & 5 deletions blueman/plugins/applet/TransferService.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@
from blueman.bluez.obex.Transfer import Transfer
from blueman.bluez.obex.Session import Session
from blueman.Functions import launch
from blueman.gui.Notification import Notification, _NotificationBubble, _NotificationDialog
from blueman.gui.Notification import Notification, _NotificationBubble, _NotificationDialog, NotificationAction
from blueman.main.Applet import BluemanApplet
from blueman.main.DbusService import DbusService, DbusError
from blueman.plugins.AppletPlugin import AppletPlugin
Expand Down Expand Up @@ -124,12 +124,17 @@ def _remove() -> bool:

# This device was neither allowed nor is it trusted -> ask for confirmation
if address not in self._allowed_devices and not (self._config['opp-accept'] and trusted):
actions = [
NotificationAction("accept", _("Accept"), on_action),
NotificationAction("reject", _("Reject"), on_action)
]
self._notification = notification = Notification(
_("Incoming file over Bluetooth"),
_("Incoming file %(0)s from %(1)s") % {"0": "<b>" + escape(filename) + "</b>",
"1": "<b>" + escape(name) + "</b>"},
30000,
actions=[("accept", _("Accept")), ("reject", _("Reject"))], actions_cb=on_action, icon_name="blueman"
actions=actions,
icon_name="blueman"
)
notification.show()
# Device is trusted or was already allowed, larger file -> display a notification, but auto-accept
Expand Down Expand Up @@ -178,12 +183,12 @@ def on_reset(_action: str) -> None:
share_path, invalid_share_path = self._make_share_path()

if invalid_share_path:
action = NotificationAction("reset", _("Reset to default"), on_reset)
text = _('Configured directory for incoming files does not exist')
secondary_text = _('Please make sure that directory "<b>%s</b>" exists or '
'configure it with blueman-services. Until then the default "%s" will be used')
self._notification = Notification(text, secondary_text % (self._config["shared-path"], share_path),
icon_name='blueman', timeout=30000,
actions=[('reset', 'Reset to default')], actions_cb=on_reset)
icon_name='blueman', timeout=30000, actions=[action])
self._notification.show()

self._watch = Manager.watch_name_owner(self._on_dbus_name_appeared, self._on_dbus_name_vanished)
Expand Down Expand Up @@ -280,7 +285,8 @@ def on_open(_action: str) -> None:
logging.info("open")
launch("xdg-open", paths=[path.as_posix()], system=True)

n.add_action("open", name, on_open)
action = NotificationAction("open", name, on_open)
n.add_action(action)

def _on_transfer_completed(self, _manager: Manager, transfer_path: ObjectPath, success: bool) -> None:
if not self._agent or transfer_path not in self._agent.transfers:
Expand Down