added empty modules and added the notifications plugin
This commit is contained in:
455
qtile/plugins/notifications.py
Normal file
455
qtile/plugins/notifications.py
Normal file
@@ -0,0 +1,455 @@
|
||||
"""
|
||||
Qtile plugin that acts as a notification server and draws notification windows.
|
||||
|
||||
Clicking on a notification will trigger the default action, e.g. telling Firefox to open
|
||||
the tab that sent the notification. If you want access to a notification's non-default
|
||||
actions then you need to disable the "actions" capability of the `Notifier` by passing
|
||||
`actions=False`.
|
||||
|
||||
Usage:
|
||||
|
||||
from graphical_notifications import Notifier
|
||||
|
||||
notifier = Notifier()
|
||||
|
||||
keys.extend([
|
||||
Key([mod], 'grave', lazy.function(notifier.prev)),
|
||||
Key([mod, 'shift'], 'grave', lazy.function(notifier.next)),
|
||||
Key(['control'], 'space', lazy.function(notifier.close)),
|
||||
])
|
||||
|
||||
Qtile versions known to work: 0.17 - 0.18
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from libqtile import configurable, hook, images, pangocffi, qtile
|
||||
from libqtile.lazy import lazy
|
||||
from libqtile.log_utils import logger
|
||||
from libqtile.notify import notifier, ClosedReason
|
||||
from libqtile.popup import Popup
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from typing import Any, Callable, Dict, List, Optional, Tuple
|
||||
|
||||
from cairocffi import ImageSurface
|
||||
|
||||
from libqtile.core.manager import Qtile
|
||||
try:
|
||||
from libqtile.notify import Notification
|
||||
except ImportError: # no dbus_next
|
||||
Notification = Any # type: ignore
|
||||
|
||||
|
||||
class Notifier(configurable.Configurable):
|
||||
"""
|
||||
This class provides a full graphical notification manager for the
|
||||
org.freedesktop.Notifications service implemented in libqtile.notify.
|
||||
|
||||
The format option determines what text is shown on the popup windows, and supports
|
||||
markup and new line characters e.g. '<b>{summary}</b>\n{body}'. Available
|
||||
placeholders are summary, body and app_name.
|
||||
|
||||
Foreground and background colours can be specified either as tuples/lists of 3
|
||||
strings, corresponding to low, normal and critical urgencies, or just a single
|
||||
string which will then be used for all urgencies. The timeout and border options can
|
||||
be set in the same way.
|
||||
|
||||
The max_windows option limits how many popup windows can be drawn at a time. When
|
||||
more notifications are recieved while the maximum number are already drawn,
|
||||
notifications are queued and displayed when existing notifications are closed.
|
||||
|
||||
TODO:
|
||||
- text overflow
|
||||
- select screen / follow mouse/keyboard focus
|
||||
- critical notifications to replace any visible non-critical notifs immediately?
|
||||
- hints: image-path, desktop-entry (for icon)
|
||||
- hints: Notifier parameters set for single notification?
|
||||
- hints: progress value e.g. int:value:42 with drawing
|
||||
|
||||
"""
|
||||
defaults = [
|
||||
('x', 32, 'x position on screen to start drawing notifications.'),
|
||||
('y', 64, 'y position on screen to start drawing notifications.'),
|
||||
('width', 192, 'Width of notifications.'),
|
||||
('height', 64, 'Height of notifications.'),
|
||||
('format', '{summary}\n{body}', 'Text format.'),
|
||||
(
|
||||
'foreground',
|
||||
('#ffffff', '#ffffff', '#ffffff'),
|
||||
'Foreground colour of notifications, in ascending order of urgency.',
|
||||
),
|
||||
(
|
||||
'background',
|
||||
('#111111', '#111111', '#111111'),
|
||||
'Background colour of notifications, in ascending order of urgency.',
|
||||
),
|
||||
(
|
||||
'border',
|
||||
('#111111', '#111111', '#111111'),
|
||||
'Border colours in ascending order of urgency. Or None for none.',
|
||||
),
|
||||
(
|
||||
'timeout',
|
||||
(5000, 5000, 0),
|
||||
'Millisecond timeout duration, in ascending order of urgency.',
|
||||
),
|
||||
('opacity', 1.0, 'Opacity of notifications.'),
|
||||
('border_width', 4, 'Line width of drawn borders.'),
|
||||
('corner_radius', None, 'Corner radius for round corners, or None.'),
|
||||
('font', 'sans', 'Font used in notifications.'),
|
||||
('fontsize', 14, 'Size of font.'),
|
||||
('fontshadow', None, 'Color for text shadows, or None for no shadows.'),
|
||||
('text_alignment', 'left', 'Text alignment: left, center or right.'),
|
||||
('horizontal_padding', None, 'Padding at sides of text.'),
|
||||
('vertical_padding', None, 'Padding at top and bottom of text.'),
|
||||
('line_spacing', 4, 'Space between lines.'),
|
||||
(
|
||||
'overflow',
|
||||
'truncate',
|
||||
'How to deal with too much text: more_width, more_height, or truncate.',
|
||||
),
|
||||
('max_windows', 2, 'Maximum number of windows to show at once.'),
|
||||
('gap', 12, 'Vertical gap between popup windows.'),
|
||||
('sticky_history', True, 'Disable timeout when browsing history.'),
|
||||
('icon_size', 36, 'Pixel size of any icons.'),
|
||||
('fullscreen', 'show', 'What to do when in fullscreen: show, hide, or queue.'),
|
||||
('screen', 'focus', 'How to select a screen: focus, mouse, or an int.'),
|
||||
('actions', True, 'Whether to enable the actions capability.'),
|
||||
]
|
||||
capabilities = {'body', 'body-markup', 'actions'}
|
||||
# specification: https://developer.gnome.org/notification-spec/
|
||||
|
||||
def __init__(self, **config) -> None:
|
||||
configurable.Configurable.__init__(self, **config)
|
||||
self.add_defaults(Notifier.defaults)
|
||||
self._hidden: List[Popup] = []
|
||||
self._shown: List[Popup] = []
|
||||
self._queue: List[Notification] = []
|
||||
self._positions: List[Tuple[int, int]] = []
|
||||
self._scroll_popup: Optional[Popup] = None
|
||||
self._current_id: int = 0
|
||||
self._notif_id: Optional[int] = None
|
||||
self._paused: bool = False
|
||||
self._icons: Dict[str, Tuple[ImageSurface, int]] = {}
|
||||
|
||||
self._make_attr_list('foreground')
|
||||
self._make_attr_list('background')
|
||||
self._make_attr_list('timeout')
|
||||
self._make_attr_list('border')
|
||||
|
||||
hook.subscribe.startup(lambda: asyncio.create_task(self._configure()))
|
||||
|
||||
if self.actions is False:
|
||||
Notifier.capabilities.remove("actions")
|
||||
|
||||
def _make_attr_list(self, attr: str) -> None:
|
||||
"""
|
||||
Turns '#000000' into ('#000000', '#000000', '#000000')
|
||||
"""
|
||||
value = getattr(self, attr)
|
||||
if not isinstance(value, (tuple, list)):
|
||||
setattr(self, attr, (value,) * 3)
|
||||
|
||||
async def _configure(self) -> None:
|
||||
"""
|
||||
This method needs to be called to set up the Notifier with the Qtile manager and
|
||||
create the required popup windows.
|
||||
"""
|
||||
if self.horizontal_padding is None:
|
||||
self.horizontal_padding = self.fontsize // 2
|
||||
if self.vertical_padding is None:
|
||||
self.vertical_padding = self.fontsize // 2
|
||||
|
||||
popup_config = {
|
||||
"x": self.x,
|
||||
"y": self.y,
|
||||
"width": self.width,
|
||||
"height": self.height,
|
||||
}
|
||||
|
||||
for opt in Popup.defaults:
|
||||
key = opt[0]
|
||||
if hasattr(self, key):
|
||||
value = getattr(self, key)
|
||||
if isinstance(value, (tuple, list)):
|
||||
popup_config[key] = value[1]
|
||||
else:
|
||||
popup_config[key] = value
|
||||
|
||||
for win in range(self.max_windows):
|
||||
popup = Popup(qtile, **popup_config)
|
||||
popup.win.process_button_click = self._process_button_click(popup)
|
||||
popup.notif = None
|
||||
self._hidden.append(popup)
|
||||
self._positions.append(
|
||||
(
|
||||
self.x,
|
||||
self.y + win * (self.height + 2 * self.border_width + self.gap)
|
||||
)
|
||||
)
|
||||
|
||||
# Clear defunct callbacks left when reloading the config
|
||||
notifier.callbacks.clear()
|
||||
notifier.close_callbacks.clear()
|
||||
|
||||
await notifier.register(
|
||||
self._notify, Notifier.capabilities, on_close=self._on_close
|
||||
)
|
||||
logger.info("Notification server started up successfully")
|
||||
|
||||
def _process_button_click(self, popup: Popup) -> Callable:
|
||||
def _(x: int, y: int, button: int) -> None:
|
||||
if button == 1:
|
||||
self._act(popup)
|
||||
self._close(popup, reason=ClosedReason.dismissed)
|
||||
if button == 3:
|
||||
self._close(popup, reason=ClosedReason.dismissed)
|
||||
return _
|
||||
|
||||
def _notify(self, notif: Notification) -> None:
|
||||
"""
|
||||
This method is registered with the NotificationManager to handle notifications
|
||||
received via dbus. They will either be drawn now or queued to be drawn soon.
|
||||
"""
|
||||
if self._paused:
|
||||
self._queue.append(notif)
|
||||
return
|
||||
|
||||
if qtile.current_window and qtile.current_window.fullscreen:
|
||||
if self.fullscreen != 'show':
|
||||
if self.fullscreen == 'queue':
|
||||
if self._unfullscreen not in hook.subscriptions:
|
||||
hook.subscribe.float_change(self._unfullscreen)
|
||||
self._queue.append(notif)
|
||||
return
|
||||
|
||||
if notif.replaces_id:
|
||||
for popup in self._shown:
|
||||
if notif.replaces_id == popup.notif.replaces_id:
|
||||
self._shown.remove(popup)
|
||||
self._send(notif, popup)
|
||||
self._reposition()
|
||||
return
|
||||
|
||||
if self._hidden:
|
||||
self._send(notif, self._hidden.pop())
|
||||
else:
|
||||
self._queue.append(notif)
|
||||
|
||||
def _on_close(self, nid: int) -> None:
|
||||
for popup in self._shown:
|
||||
self._close(popup, nid=nid, reason=ClosedReason.method)
|
||||
|
||||
def _unfullscreen(self) -> None:
|
||||
"""
|
||||
Begin displaying of queue notifications after leaving fullscreen.
|
||||
"""
|
||||
if not qtile.current_window.fullscreen:
|
||||
hook.unsubscribe.float_change(self._unfullscreen)
|
||||
self._renotify()
|
||||
|
||||
def _renotify(self) -> None:
|
||||
"""
|
||||
If we hold off temporarily on sending notifications and accumulate a queue, we
|
||||
should use this to send the queue through self._notify again.
|
||||
"""
|
||||
queue = self._queue.copy()
|
||||
self._queue.clear()
|
||||
while queue:
|
||||
self._notify(queue.pop(0))
|
||||
|
||||
def _send(
|
||||
self, notif: Notification, popup: Popup, timeout: Optional[int] = None
|
||||
) -> None:
|
||||
"""
|
||||
Draw the desired notification using the specified Popup instance.
|
||||
"""
|
||||
text = self._get_text(notif)
|
||||
|
||||
if "urgency" in notif.hints:
|
||||
urgency = notif.hints["urgency"].value
|
||||
else:
|
||||
urgency = 1
|
||||
|
||||
self._current_id += 1
|
||||
popup.id = self._current_id # Used for closing the popup
|
||||
popup.notif = notif # Used for finding the visible popup's notif for actions
|
||||
if popup not in self._shown:
|
||||
self._shown.append(popup)
|
||||
popup.x, popup.y = self._get_coordinates()
|
||||
popup.place()
|
||||
icon = self._load_icon(notif)
|
||||
popup.unhide()
|
||||
|
||||
popup.background = self.background[urgency]
|
||||
popup.layout.colour = self.foreground[urgency]
|
||||
popup.clear()
|
||||
|
||||
if icon:
|
||||
popup.draw_image(
|
||||
icon[0],
|
||||
self.horizontal_padding,
|
||||
1 + (self.height - icon[1]) / 2,
|
||||
)
|
||||
popup.horizontal_padding += self.icon_size + self.horizontal_padding / 2
|
||||
|
||||
for num, line in enumerate(text.split('\n')):
|
||||
popup.layout.text = line
|
||||
y = self.vertical_padding + num * (popup.layout.height + self.line_spacing)
|
||||
popup.draw_text(y=y)
|
||||
if self.border_width:
|
||||
popup.set_border(self.border[urgency])
|
||||
popup.draw()
|
||||
if icon:
|
||||
popup.horizontal_padding = self.horizontal_padding
|
||||
|
||||
if timeout is None:
|
||||
if notif.timeout is None or notif.timeout < 0:
|
||||
timeout = self.timeout[urgency]
|
||||
else:
|
||||
timeout = notif.timeout
|
||||
elif timeout < 0:
|
||||
timeout = self.timeout[urgency]
|
||||
if timeout > 0:
|
||||
qtile.call_later(timeout / 1000, self._close, popup, self._current_id)
|
||||
|
||||
def _get_text(self, notif: Notification) -> str:
|
||||
summary = ''
|
||||
body = ''
|
||||
app_name = ''
|
||||
if notif.summary:
|
||||
summary = pangocffi.markup_escape_text(notif.summary)
|
||||
if notif.body:
|
||||
body = pangocffi.markup_escape_text(notif.body)
|
||||
if notif.app_name:
|
||||
app_name = pangocffi.markup_escape_text(notif.app_name)
|
||||
return self.format.format(summary=summary, body=body, app_name=app_name)
|
||||
|
||||
def _get_coordinates(self) -> Tuple[int, int]:
|
||||
x, y = self._positions[len(self._shown) - 1]
|
||||
if isinstance(self.screen, int):
|
||||
screen = qtile.screens[self.screen]
|
||||
elif self.screen == 'focus':
|
||||
screen = qtile.current_screen
|
||||
elif self.screen == 'mouse':
|
||||
screen = qtile.find_screen(*qtile.mouse_position)
|
||||
return x + screen.x, y + screen.y
|
||||
|
||||
def _close(self, popup: Popup, nid: Optional[int] = None, reason=1) -> None:
|
||||
"""
|
||||
Close the specified Popup instance.
|
||||
"""
|
||||
if popup in self._shown:
|
||||
if nid is not None and popup.id != nid:
|
||||
return
|
||||
self._shown.remove(popup)
|
||||
if self._scroll_popup is popup:
|
||||
self._scroll_popup = None
|
||||
self._notif_id = None
|
||||
popup.hide()
|
||||
if self._queue and not self._paused:
|
||||
self._send(self._queue.pop(0), popup)
|
||||
else:
|
||||
self._hidden.append(popup)
|
||||
notifier._service.NotificationClosed(popup.notif.id, reason)
|
||||
self._reposition()
|
||||
|
||||
def _act(self, popup: Popup) -> None:
|
||||
"""
|
||||
Execute the actions specified by the notification visible on a clicked popup.
|
||||
"""
|
||||
# Currently this always invokes default action
|
||||
#actions = {i: l for i, l in zip(notif.actions[:-1:2], notif.actions[1::2])}
|
||||
if popup.notif.actions:
|
||||
notifier._service.ActionInvoked(popup.notif.id, popup.notif.actions[0])
|
||||
|
||||
def _reposition(self) -> None:
|
||||
for index, shown in enumerate(self._shown):
|
||||
shown.x, shown.y = self._positions[index]
|
||||
shown.place()
|
||||
|
||||
def _load_icon(self, notif: Notification) -> Optional[Tuple[ImageSurface, int]]:
|
||||
if not notif.app_icon:
|
||||
return None
|
||||
if notif.app_icon in self._icons:
|
||||
return self._icons.get(notif.app_icon)
|
||||
try:
|
||||
img = images.Img.from_path(notif.app_icon)
|
||||
if img.width > img.height:
|
||||
img.resize(width=self.icon_size)
|
||||
else:
|
||||
img.resize(height=self.icon_size)
|
||||
surface, _ = images._decode_to_image_surface(
|
||||
img.bytes_img, img.width, img.height
|
||||
)
|
||||
self._icons[notif.app_icon] = surface, surface.get_height()
|
||||
except (FileNotFoundError, images.LoadingError, IsADirectoryError) as e:
|
||||
logger.exception(e)
|
||||
self._icons[notif.app_icon] = None
|
||||
return self._icons[notif.app_icon]
|
||||
|
||||
def close(self, _qtile=None) -> None:
|
||||
"""
|
||||
Close the oldest of all visible popup windows.
|
||||
"""
|
||||
if self._shown:
|
||||
self._close(self._shown[0])
|
||||
|
||||
def close_all(self, _qtile=None) -> None:
|
||||
"""
|
||||
Close all popup windows.
|
||||
"""
|
||||
self._queue.clear()
|
||||
while self._shown:
|
||||
self._close(self._shown[0])
|
||||
|
||||
def prev(self, _qtile=None) -> None:
|
||||
"""
|
||||
Display the previous notification in the history.
|
||||
"""
|
||||
if notifier.notifications:
|
||||
if self._scroll_popup is None:
|
||||
if self._hidden:
|
||||
self._scroll_popup = self._hidden.pop(0)
|
||||
else:
|
||||
self._scroll_popup = self._shown[0]
|
||||
self._notif_id = len(notifier.notifications)
|
||||
if self._notif_id > 0:
|
||||
self._notif_id -= 1
|
||||
self._send(
|
||||
notifier.notifications[self._notif_id],
|
||||
self._scroll_popup,
|
||||
0 if self.sticky_history else None,
|
||||
)
|
||||
|
||||
def next(self, _qtile=None) -> None:
|
||||
"""
|
||||
Display the next notification in the history.
|
||||
"""
|
||||
if self._scroll_popup:
|
||||
if self._notif_id < len(notifier.notifications) - 1:
|
||||
self._notif_id += 1
|
||||
if self._scroll_popup in self._shown:
|
||||
self._shown.remove(self._scroll_popup)
|
||||
self._send(
|
||||
notifier.notifications[self._notif_id],
|
||||
self._scroll_popup,
|
||||
0 if self.sticky_history else None,
|
||||
)
|
||||
|
||||
def pause(self, _qtile=None) -> None:
|
||||
"""
|
||||
Pause display of notifications on screen. Notifications will be queued and
|
||||
presented as usual when this is called again.
|
||||
"""
|
||||
if self._paused:
|
||||
self._paused = False
|
||||
self._renotify()
|
||||
else:
|
||||
self._paused = True
|
||||
while self._shown:
|
||||
self._close(self._shown[0])
|
||||
Reference in New Issue
Block a user