diff --git a/docs/desktop-notifications.rst b/docs/desktop-notifications.rst index f8d3deb8d..012a845bb 100644 --- a/docs/desktop-notifications.rst +++ b/docs/desktop-notifications.rst @@ -171,6 +171,20 @@ This will close a previous notification with the specified id. If no such notification exists (perhaps because it was already closed or it was activated) then the request is ignored. + +Automatically expiring notifications +------------------------------------- + +A notification can be marked as expiring (being closed) automatically after +a specified number of milliseconds using the ``w`` key. The default if +unspecified is ``-1`` which means to use whatever expiry policy the OS has for +notifications. A value of ``0`` means the notification should never expire. +Values greater than zero specify the number of milliseconds after which the +notification should be auto-closed. Note that the value of ``0`` +is best effort, some platforms honor it and some do not. Positive values +are robust, since they can be implemented by the terminal emulator itself, +by manually closing the notification after the expiry time. + .. _notifications_query: Querying for support @@ -215,6 +229,8 @@ Key Value ``c`` ``c=1`` if the terminal supports close events, otherwise the ``c`` must be omitted. + +``w`` ``w=1`` if the terminal supports auto expiring of notifications. ======= ================================================================================ In the future, if this protocol expands, more keys might be added. Clients must @@ -274,6 +290,8 @@ Key Value Default Description ``t`` :rfc:`base64 <4648>` ``unset`` The type of the notification. Can be used to filter out notifications. encoded UTF-8 notification type + +``w`` ``>=-1`` ``-1`` The number of milliseconds to auto-close the notification after. ======= ==================== ========== ================= diff --git a/kitty/notifications.py b/kitty/notifications.py index a875eb365..1f99da77d 100644 --- a/kitty/notifications.py +++ b/kitty/notifications.py @@ -6,12 +6,13 @@ import re from collections import OrderedDict from contextlib import suppress from enum import Enum +from functools import partial from itertools import count from typing import Any, Callable, Dict, FrozenSet, Iterator, List, NamedTuple, Optional, Sequence, Set, Tuple, Union from weakref import ReferenceType, ref from .constants import cache_dir, config_dir, is_macos, logo_png_file -from .fast_data_types import ESC_OSC, StreamingBase64Decoder, base64_decode, current_focused_os_window_id, get_boss, get_options +from .fast_data_types import ESC_OSC, StreamingBase64Decoder, add_timer, base64_decode, current_focused_os_window_id, get_boss, get_options from .types import run_once from .typing import WindowType from .utils import get_custom_window_icon, log_error, sanitize_control_codes @@ -201,6 +202,7 @@ class NotificationCommand: icon_name: str = '' application_name: str = '' notification_type: str = '' + timeout: int = -2 # event callbacks on_activation: Optional[Callable[['NotificationCommand'], None]] = None @@ -297,6 +299,11 @@ class NotificationCommand: self.notification_type = base64_decode(v).decode('utf-8', 'replace') except Exception: self.log('Ignoring invalid notification type in notification: {v!r}') + elif k == 'w': + try: + self.timeout = max(-1, int(v)) + except Exception: + self.log('Ignoring invalid timeout in notification: {v!r}') if not prev.done and prev.identifier == self.identifier: self.merge_metadata(prev) return payload_type, payload_is_encoded @@ -319,6 +326,8 @@ class NotificationCommand: self.application_name = prev.application_name if not self.notification_type: self.notification_type = prev.notification_type + if self.timeout < -1: + self.timeout = prev.timeout self.icon_path = prev.icon_path def create_payload_buffer(self, payload_type: PayloadType) -> EncodedDataStore: @@ -380,6 +389,7 @@ class NotificationCommand: self.body = '' self.urgency = Urgency.Normal if self.urgency is None else self.urgency self.close_response_requested = bool(self.close_response_requested) + self.timeout = max(-1, self.timeout) def matches_rule_item(self, location:str, query:str) -> bool: import re @@ -431,7 +441,7 @@ class DesktopIntegration: i = f'i={identifier or "0"}:' p = ','.join(x.value for x in PayloadType if x.value) c = ':c=1' if self.supports_close_events else '' - return f'99;{i}p=?;a={actions}:o={when}:u={urgency}:p={p}{c}' + return f'99;{i}p=?;a={actions}:o={when}:u={urgency}:p={p}{c}:w=1' class MacOSIntegration(DesktopIntegration): @@ -473,15 +483,17 @@ class MacOSIntegration(DesktopIntegration): return desktop_notification_id def notification_activated(self, event: str, ident: str) -> None: - if debug_desktop_integration: - log_error(f'Notification {ident} {event=}') if event == 'live': + if debug_desktop_integration: + log_error('Got list of live notifications:', ident) live_ids = tuple(int(x) for x in ident.split(',') if x) self.notification_manager.purge_dead_notifications(live_ids) self.live_notification_queries, queries = [], self.live_notification_queries for channel_id, req_id in queries: self.notification_manager.send_live_response(channel_id, req_id, live_ids) return + if debug_desktop_integration: + log_error(f'Notification {ident} {event=}') try: desktop_notification_id = int(ident) except Exception: @@ -564,8 +576,8 @@ class FreeDesktopIntegration(DesktopIntegration): if existing_desktop_notification_id: replaces_dbus_id = self.get_dbus_notification_id(existing_desktop_notification_id, 'notify') or 0 desktop_notification_id = dbus_send_notification( - app_name=nc.application_name or 'kitty', app_icon=app_icon, title=nc.title, body=body, timeout=-1, urgency=nc.urgency.value, - replaces=replaces_dbus_id) + app_name=nc.application_name or 'kitty', app_icon=app_icon, title=nc.title, body=body, timeout=nc.timeout, + urgency=nc.urgency.value, replaces=replaces_dbus_id) if debug_desktop_integration: log_error(f'Requested creation of notification with {desktop_notification_id=}') if existing_desktop_notification_id and replaces_dbus_id: @@ -667,6 +679,8 @@ class NotificationManager: def notification_created(self, desktop_notification_id: int) -> None: if n := self.in_progress_notification_commands.get(desktop_notification_id): n.created_by_desktop = True + if n.timeout > 0: + add_timer(partial(self.expire_notification, desktop_notification_id, id(n)), n.timeout / 1000, False) def notification_activation_token_received(self, desktop_notification_id: int, token: str) -> None: if n := self.in_progress_notification_commands.get(desktop_notification_id): @@ -768,6 +782,11 @@ class NotificationManager: self.send_closed_response(channel_id, cmd.identifier, untracked=True) return desktop_notification_id + def expire_notification(self, desktop_notification_id: int, command_id: int, timer_id: int) -> None: + if n := self.in_progress_notification_commands.get(desktop_notification_id): + if id(n) == command_id: + self.desktop_integration.close_notification(desktop_notification_id) + def register_in_progress_notification(self, cmd: NotificationCommand, desktop_notification_id: int) -> None: cmd.desktop_notification_id = desktop_notification_id self.in_progress_notification_commands[desktop_notification_id] = cmd diff --git a/kitty_tests/notifications.py b/kitty_tests/notifications.py index c0de39d3c..897ee7925 100644 --- a/kitty_tests/notifications.py +++ b/kitty_tests/notifications.py @@ -12,10 +12,13 @@ from kitty.notifications import Channel, DesktopIntegration, IconDataCache, Noti from . import BaseTest -def n(title='title', body='', urgency=Urgency.Normal, desktop_notification_id=1, icon_name='', icon_path='', application_name='', notification_type=''): +def n( + title='title', body='', urgency=Urgency.Normal, desktop_notification_id=1, icon_name='', icon_path='', + application_name='', notification_type='', timeout=-1, +): return { 'title': title, 'body': body, 'urgency': urgency, 'id': desktop_notification_id, 'icon_name': icon_name, 'icon_path': icon_path, - 'application_name': application_name, 'notification_type': notification_type, + 'application_name': application_name, 'notification_type': notification_type, 'timeout': timeout } @@ -50,7 +53,7 @@ class DesktopIntegration(DesktopIntegration): self.counter += 1 did = self.counter title, body, urgency = cmd.title, cmd.body, (Urgency.Normal if cmd.urgency is None else cmd.urgency) - ans = n(title, body, urgency, did, cmd.icon_name, os.path.basename(cmd.icon_path), cmd.application_name, cmd.notification_type) + ans = n(title, body, urgency, did, cmd.icon_name, os.path.basename(cmd.icon_path), cmd.application_name, cmd.notification_type, timeout=cmd.timeout) self.notifications.append(ans) return self.counter @@ -235,7 +238,7 @@ def do_test(self: 'TestNotifications', tdir: str) -> None: # Test querying h('i=xyz:p=?') self.assertFalse(di.notifications) - qr = 'a=focus,report:o=always,unfocused,invisible:u=0,1,2:p=title,body,?,close,icon,alive:c=1' + qr = 'a=focus,report:o=always,unfocused,invisible:u=0,1,2:p=title,body,?,close,icon,alive:c=1:w=1' self.ae(ch.responses, [f'99;i=xyz:p=?;{qr}']) reset() h('p=?') @@ -266,6 +269,11 @@ def do_test(self: 'TestNotifications', tdir: str) -> None: self.ae(di.notifications, [n(application_name='app', notification_type='test')]) reset() + # Test timeout + h('w=3;title') + self.ae(di.notifications, [n(timeout=3)]) + reset() + # Test Disk Cache dc = IconDataCache(base_cache_dir=tdir, max_cache_size=4) cache_dir = dc._ensure_state()