diff --git a/config/assets/launcher.json b/config/assets/launcher.json index 6c4313d..223a68f 100644 --- a/config/assets/launcher.json +++ b/config/assets/launcher.json @@ -45,8 +45,7 @@ }, "kanban": { "icon": "kanban", - "description": "Kanban - Task management with kanban-style columns", - "examples": ["kanban add Fix bug", "kanban move Fix bug", "kanban done Fix bug"] + "description": "Kanban - Task management with kanban-style columns" }, "cal": { "icon": "calendar", @@ -68,11 +67,6 @@ "icon": "auth", "description": "Manage TOTP codes and 2FA authentication" }, - "notif": { - "icon": "notifications", - "description": "Notification History - Search and manage notification history", - "examples": ["notif", "notif clear", "notif search term"] - }, "pass": { "icon": "key", "description": "Password Manager - Search and manage passwords" diff --git a/config/hypr/colors.conf b/config/hypr/colors.conf index fbe8a2b..6269b38 100644 --- a/config/hypr/colors.conf +++ b/config/hypr/colors.conf @@ -1,13 +1,13 @@ -$wallpaper = /home/sankalp/Pictures/wallpapers/91.jpg -$background = 0e1416 -$foreground = dee3e5 +$wallpaper = /home/sankalp/Pictures/wallpapers/97.jpg +$background = 101417 +$foreground = dfe3e7 -$primary = 82d3e1 -$secondary = b1cbd0 -$tertiary = bbc6ea -$surface = 0e1416 -$surface_bright = 343a3b -$outline = 899294 +$primary = 93cdf6 +$secondary = b6c9d8 +$tertiary = cdc0e9 +$surface = 101417 +$surface_bright = 353a3d +$outline = 8b9198 $error = ff5945 $shadow = 000000 @@ -17,11 +17,11 @@ $green = 95d5a7 $yellow = b8cf84 $blue = afc6ff $magenta = e4b7f3 -$cyan = 80d4db +$cyan = 82d3e2 $white = 82d3e0 $red_dim = ff4c5c $green_dim = 4cb86a $yellow_dim = 8baa43 $blue_dim = 497dff $magenta_dim = c360e4 -$cyan_dim = 36b4bf +$cyan_dim = 30b5ce diff --git a/example_notifications.py b/example_notifications.py new file mode 100644 index 0000000..4493e68 --- /dev/null +++ b/example_notifications.py @@ -0,0 +1,1229 @@ +import json +import locale +import os +import uuid +from datetime import datetime, timedelta + +from fabric.notifications.service import (Notification, NotificationAction, + Notifications) +from fabric.widgets.box import Box +from fabric.widgets.button import Button +from fabric.widgets.centerbox import CenterBox +from fabric.widgets.image import Image +from fabric.widgets.label import Label +from fabric.widgets.revealer import Revealer +from fabric.widgets.scrolledwindow import ScrolledWindow +from gi.repository import GdkPixbuf, GLib, Gtk +from loguru import logger + +import config.data as data +import modules.icons as icons +from widgets.rounded_image import CustomImage +from widgets.wayland import WaylandWindow as Window + +PERSISTENT_DIR = f"/tmp/{data.APP_NAME}/notifications" +PERSISTENT_HISTORY_FILE = os.path.join(PERSISTENT_DIR, "notification_history.json") + +def cache_notification_pixbuf(notification_box): + """ + Saves a scaled pixbuf (48x48) in the cache directory and returns the cache file path. + """ + notification = notification_box.notification + if notification.image_pixbuf: + os.makedirs(PERSISTENT_DIR, exist_ok=True) + cache_file = os.path.join(PERSISTENT_DIR, f"notification_{notification_box.uuid}.png") + logger.debug(f"Caching image for notification {notification.id} to: {cache_file}") + try: + scaled = notification.image_pixbuf.scale_simple(48, 48, GdkPixbuf.InterpType.BILINEAR) + scaled.savev(cache_file, "png", [], []) + logger.info(f"Successfully cached image for notification {notification.id} to: {cache_file}") + return cache_file + except Exception as e: + logger.error(f"Error caching image for notification {notification.id}: {e}") + return None + else: + logger.debug(f"Notification {notification.id} has no image_pixbuf to cache.") + return None + +def load_scaled_pixbuf(notification_box, width, height): + """ + Loads and scales a pixbuf for a notification_box, prioritizing cached images. + """ + notification = notification_box.notification + if not hasattr(notification_box, 'notification') or notification is None: + logger.error("load_scaled_pixbuf: notification_box.notification is None or not set!") + return None + + pixbuf = None + if hasattr(notification_box, "cached_image_path") and notification_box.cached_image_path and os.path.exists(notification_box.cached_image_path): + try: + logger.debug(f"Attempting to load cached image from: {notification_box.cached_image_path} for notification {notification.id}") + pixbuf = GdkPixbuf.Pixbuf.new_from_file(notification_box.cached_image_path) + if pixbuf: + pixbuf = pixbuf.scale_simple(width, height, GdkPixbuf.InterpType.BILINEAR) + logger.info(f"Successfully loaded cached image from: {notification_box.cached_image_path} for notification {notification.id}") + return pixbuf + except Exception as e: + logger.error(f"Error loading cached image from {notification_box.cached_image_path} for notification {notification.id}: {e}") + logger.warning(f"Falling back to notification.image_pixbuf for notification {notification.id}") + + if notification.image_pixbuf: + logger.debug(f"Loading image directly from notification.image_pixbuf for notification {notification.id}") + pixbuf = notification.image_pixbuf.scale_simple(width, height, GdkPixbuf.InterpType.BILINEAR) + return pixbuf + + logger.debug(f"No image_pixbuf or cached image found, trying app icon for notification {notification.id}") + return get_app_icon_pixbuf(notification.app_icon, width, height) + +def get_app_icon_pixbuf(icon_path, width, height): + """ + Loads and scales a pixbuf from an app icon path. + """ + if not icon_path: + return None + if icon_path.startswith("file://"): + icon_path = icon_path[7:] + if not os.path.exists(icon_path): + logger.warning(f"Icon path does not exist: {icon_path}") + return None + try: + pixbuf = GdkPixbuf.Pixbuf.new_from_file(icon_path) + return pixbuf.scale_simple(width, height, GdkPixbuf.InterpType.BILINEAR) + except Exception as e: + logger.error(f"Failed to load or scale icon: {e}") + return None + +class ActionButton(Button): + def __init__(self, action: NotificationAction, index: int, total: int, notification_box): + super().__init__( + name="action-button", + h_expand=True, + on_clicked=self.on_clicked, + child=Label(name="button-label", h_expand=True, h_align="fill", ellipsization="end", max_chars_width=1, label=action.label), + ) + self.action = action + self.notification_box = notification_box + style_class = ( + "start-action" if index == 0 + else "end-action" if index == total - 1 + else "middle-action" + ) + self.add_style_class(style_class) + self.connect("enter-notify-event", lambda *_: notification_box.hover_button(self)) + self.connect("leave-notify-event", lambda *_: notification_box.unhover_button(self)) + + def on_clicked(self, *_): + self.action.invoke() + self.action.parent.close("dismissed-by-user") + +class NotificationBox(Box): + def __init__(self, notification: Notification, timeout_ms=5000, **kwargs): + super().__init__( + name="notification-box", + orientation="v", + h_align="fill", + h_expand=True, + children=[], + ) + self.notification = notification + self.uuid = str(uuid.uuid4()) + + if timeout_ms == 0: + self.timeout_ms = 0 + else: + + live_timeout = getattr(self.notification, 'timeout', -1) + self.timeout_ms = live_timeout if live_timeout != -1 else timeout_ms + self._timeout_id = None + self._container = None + self.cached_image_path = None + + if self.timeout_ms > 0: + self.start_timeout() + + if self.notification.image_pixbuf: + cache_path = cache_notification_pixbuf(self) + if cache_path: + self.cached_image_path = cache_path + logger.debug(f"NotificationBox {self.uuid}: Cached image path set to: {self.cached_image_path}") + else: + logger.warning(f"NotificationBox {self.uuid}: Caching failed, cached_image_path not set.") + else: + logger.debug(f"NotificationBox {self.uuid}: No image to cache.") + + content = self.create_content() + action_buttons = self.create_action_buttons() + self.add(content) + if action_buttons: + self.add(action_buttons) + + self.connect("enter-notify-event", self.on_hover_enter) + self.connect("leave-notify-event", self.on_hover_leave) + + self._destroyed = False + self._is_history = False + logger.debug(f"NotificationBox {self.uuid} created for notification {notification.id}") + + def set_is_history(self, is_history): + self._is_history = is_history + + def set_container(self, container): + self._container = container + + def get_container(self): + return self._container + + def create_header(self): + notification = self.notification + self.app_icon_image = ( + Image( + name="notification-icon", + image_file=notification.app_icon[7:], + size=24, + ) if "file://" in notification.app_icon else + Image( + name="notification-icon", + icon_name="dialog-information-symbolic" or notification.app_icon, + icon_size=24, + ) + ) + self.app_name_label_header = Label( + notification.app_name, + name="notification-app-name", + h_align="start" + ) + self.header_close_button = self.create_close_button() + + return CenterBox( + name="notification-title", + start_children=[ + Box( + spacing=4, + children=[ + self.app_icon_image, + self.app_name_label_header, + ] + ) + ], + end_children=[self.header_close_button] + ) + + def create_content(self): + notification = self.notification + pixbuf = load_scaled_pixbuf(self, 48, 48) + self.notification_image_box = Box( + name="notification-image", + orientation="v", + children=[CustomImage(pixbuf=pixbuf), Box(v_expand=True)], + ) + self.notification_summary_label = Label( + name="notification-summary", + markup=notification.summary, + h_align="start", + max_chars_width=16, + ellipsization="end", + ) + self.notification_app_name_label_content = Label( + name="notification-app-name", + markup=notification.app_name, + h_align="start", + max_chars_width=16, + ellipsization="end", + ) + self.notification_body_label = Label( + markup=notification.body, + h_align="start", + max_chars_width=34, + ellipsization="end", + ) if notification.body else Box() + self.notification_body_label.set_single_line_mode(True) if notification.body else None + self.notification_text_box = Box( + name="notification-text", + orientation="v", + v_align="center", + h_expand=True, + h_align="start", + children=[ + Box( + name="notification-summary-box", + orientation="h", + children=[ + self.notification_summary_label, + Box(name="notif-sep", h_expand=False, v_expand=False, h_align="center", v_align="center"), + self.notification_app_name_label_content, + ], + ), + self.notification_body_label, + ], + ) + self.content_close_button = self.create_close_button() + self.content_close_button_box = Box( + orientation="v", + children=[ + self.content_close_button, + ], + ) + + return Box( + name="notification-content", + spacing=8, + children=[ + self.notification_image_box, + self.notification_text_box, + self.content_close_button_box, + ], + ) + + def create_action_buttons(self): + notification = self.notification + if not notification.actions: + return None + + grid = Gtk.Grid() + grid.set_column_homogeneous(True) + grid.set_column_spacing(4) + for i, action in enumerate(notification.actions): + action_button = ActionButton(action, i, len(notification.actions), self) + grid.attach(action_button, i, 0, 1, 1) + return grid + + def create_close_button(self): + self.close_button = Button( + name="notif-close-button", + child=Label(name="notif-close-label", markup=icons.cancel), + on_clicked=lambda *_: self.notification.close("dismissed-by-user"), + ) + self.close_button.connect("enter-notify-event", lambda *_: self.hover_button(self.close_button)) + self.close_button.connect("leave-notify-event", lambda *_: self.unhover_button(self.close_button)) + return self.close_button + + def on_hover_enter(self, *args): + if self._container: + self._container.pause_and_reset_all_timeouts() + + def on_hover_leave(self, *args): + if self._container: + self._container.resume_all_timeouts() + + def start_timeout(self): + self.stop_timeout() + self._timeout_id = GLib.timeout_add(self.timeout_ms, self.close_notification) + + def stop_timeout(self): + if self._timeout_id is not None: + GLib.source_remove(self._timeout_id) + self._timeout_id = None + + def close_notification(self): + if not self._destroyed: + try: + logger.debug(f"Notification {self.notification.id} timeout expired, closing notification.") + self.notification.close("expired") + self.stop_timeout() + except Exception as e: + logger.error(f"Error in close_notification for notification {self.notification.id}: {e}") + return False + + def destroy(self, from_history_delete=False): + logger.debug(f"NotificationBox destroy called for notification: {self.notification.id}, from_history_delete: {from_history_delete}, is_history: {self._is_history}") + if hasattr(self, "cached_image_path") and self.cached_image_path and os.path.exists(self.cached_image_path) and (not self._is_history or from_history_delete): + try: + os.remove(self.cached_image_path) + logger.info(f"Deleted cached image: {self.cached_image_path}") + except Exception as e: + logger.error(f"Error deleting cached image {self.cached_image_path}: {e}") + self._destroyed = True + self.stop_timeout() + super().destroy() + + def hover_button(self, button): + if self._container: + self._container.pause_and_reset_all_timeouts() + + def unhover_button(self, button): + if self._container: + self._container.resume_all_timeouts() + +class HistoricalNotification(object): + def __init__(self, id, app_icon, summary, body, app_name, timestamp, cached_image_path=None): + self.id = id + self.app_icon = app_icon + self.summary = summary + self.body = body + self.app_name = app_name + self.timestamp = timestamp + self.cached_image_path = cached_image_path + self.image_pixbuf = None + self.actions = [] + self.cached_scaled_pixbuf = None + +class NotificationHistory(Box): + def __init__(self, **kwargs): + super().__init__( + name="notification-history", + orientation="v", + **kwargs + ) + + self.containers = [] + self.header_label = Label( + name="nhh", + label="Notifications", + h_align="start", + h_expand=True, + ) + self.header_switch = Gtk.Switch(name="dnd-switch") + self.header_switch.set_vexpand(False) + self.header_switch.set_valign(Gtk.Align.CENTER) + self.header_switch.set_active(False) + self.header_clean = Button( + name="nhh-button", + child=Label(name="nhh-button-label", markup=icons.trash), + on_clicked=self.clear_history, + ) + self.do_not_disturb_enabled = False + self.header_switch.connect("notify::active", self.on_do_not_disturb_changed) + self.dnd_label = Label(name="dnd-label", markup=icons.notifications_off) + + self.history_header = CenterBox( + name="notification-history-header", + spacing=8, + start_children=[self.header_switch, self.dnd_label], + center_children=[self.header_label], + end_children=[self.header_clean], + ) + self.notifications_list = Box( + name="notifications-list", + orientation="v", + spacing=4, + h_expand=True, + v_expand=True, + h_align="fill", + v_align="fill", + ) + self.no_notifications_label = Label( + name="no-notif", + markup=icons.notifications_clear, + v_align="fill", + h_align="fill", + v_expand=True, + h_expand=True, + justification="center", + ) + self.no_notifications_box = Box( + name="no-notifications-box", + v_align="fill", + h_align="fill", + v_expand=True, + h_expand=True, + children=[self.no_notifications_label], + ) + self.scrolled_window = ScrolledWindow( + name="notification-history-scrolled-window", + orientation="v", + h_expand=True, + v_expand=True, + h_align="fill", + v_align="fill", + propagate_width=False, + propagate_height=False, + ) + self.scrolled_window_viewport_box = Box(orientation="v", children=[self.notifications_list, self.no_notifications_box]) + self.scrolled_window.add_with_viewport(self.scrolled_window_viewport_box) + self.persistent_notifications = [] + self.add(self.history_header) + self.add(self.scrolled_window) + self._load_persistent_history() + self._cleanup_orphan_cached_images() + self.schedule_midnight_update() + + self.LIMITED_APPS_HISTORY = ["Spotify"] + + def get_ordinal(self, n): + if 11 <= (n % 100) <= 13: + return 'th' + else: + return {1: 'st', 2: 'nd', 3: 'rd'}.get(n % 10, 'th') + + def get_date_header(self, dt): + now = datetime.now() + today = now.date() + date = dt.date() + if date == today: + return "Today" + elif date == today - timedelta(days=1): + return "Yesterday" + else: + original_locale = locale.getlocale(locale.LC_TIME) + try: + locale.setlocale(locale.LC_TIME, ('en_US', 'UTF-8')) + except locale.Error: + locale.setlocale(locale.LC_TIME, 'C') + try: + day = dt.day + ordinal = self.get_ordinal(day) + month = dt.strftime("%B") + if dt.year == now.year: + result = f"{month} {day}{ordinal}" + else: + result = f"{month} {day}{ordinal}, {dt.year}" + finally: + locale.setlocale(locale.LC_TIME, original_locale) + return result + + def schedule_midnight_update(self): + now = datetime.now() + next_midnight = datetime.combine(now.date() + timedelta(days=1), datetime.min.time()) + delta_seconds = (next_midnight - now).total_seconds() + GLib.timeout_add_seconds(int(delta_seconds), self.on_midnight) + + def on_midnight(self): + self.rebuild_with_separators() + self.schedule_midnight_update() + return GLib.SOURCE_REMOVE + + def create_date_separator(self, date_header): + return Box( + name="notif-date-sep", + children=[ + Label( + name="notif-date-sep-label", + label=date_header, + h_align="center", + h_expand=True, + ) + ] + ) + + def rebuild_with_separators(self): + GLib.idle_add(self._do_rebuild_with_separators) + + def _do_rebuild_with_separators(self): + children = list(self.notifications_list.get_children()) + for child in children: + self.notifications_list.remove(child) + + current_date_header = None + last_date_header = None + for container in sorted(self.containers, key=lambda x: x.arrival_time, reverse=True): + arrival_time = container.arrival_time + date_header = self.get_date_header(arrival_time) + if date_header != current_date_header: + sep = self.create_date_separator(date_header) + self.notifications_list.add(sep) + current_date_header = date_header + last_date_header = date_header + self.notifications_list.add(container) + + if not self.containers and last_date_header: + for child in list(self.notifications_list.get_children()): + if child.get_name() == "notif-date-sep": + self.notifications_list.remove(child) + + self.notifications_list.show_all() + self.update_no_notifications_label_visibility() + + def on_do_not_disturb_changed(self, switch, pspec): + self.do_not_disturb_enabled = switch.get_active() + logger.info(f"Do Not Disturb mode {'enabled' if self.do_not_disturb_enabled else 'disabled'}") + + def clear_history(self, *args): + for child in self.notifications_list.get_children()[:]: + container = child + notif_box = container.notification_box if hasattr(container, "notification_box") else None + if notif_box: + notif_box.destroy(from_history_delete=True) + self.notifications_list.remove(child) + child.destroy() + + if os.path.exists(PERSISTENT_HISTORY_FILE): + try: + os.remove(PERSISTENT_HISTORY_FILE) + logger.info("Notification history cleared and persistent file deleted.") + except Exception as e: + logger.error(f"Error deleting persistent history file: {e}") + self.persistent_notifications = [] + self.containers = [] + self.rebuild_with_separators() + + def _load_persistent_history(self): + if not os.path.exists(PERSISTENT_DIR): + os.makedirs(PERSISTENT_DIR, exist_ok=True) + if os.path.exists(PERSISTENT_HISTORY_FILE): + try: + with open(PERSISTENT_HISTORY_FILE, "r") as f: + self.persistent_notifications = json.load(f) + for note in reversed(self.persistent_notifications): + self._add_historical_notification(note) + except Exception as e: + logger.error(f"Error loading persistent history: {e}") + GLib.idle_add(self.update_no_notifications_label_visibility) + + def _save_persistent_history(self): + try: + with open(PERSISTENT_HISTORY_FILE, "w") as f: + json.dump(self.persistent_notifications, f) + except Exception as e: + logger.error(f"Error saving persistent history: {e}") + + def delete_historical_notification(self, note_id, container): + if hasattr(container, "notification_box"): + notif_box = container.notification_box + notif_box.destroy(from_history_delete=True) + + target_note_id_str = str(note_id) + + new_persistent_notifications = [] + removed_from_list = False + for note_in_list in self.persistent_notifications: + current_note_id_str = str(note_in_list.get("id")) + if current_note_id_str == target_note_id_str: + removed_from_list = True + + continue + new_persistent_notifications.append(note_in_list) + + if removed_from_list: + self.persistent_notifications = new_persistent_notifications + logger.info(f"Notification with ID {target_note_id_str} was marked for removal from persistent_notifications list.") + else: + logger.warning(f"Notification with ID {target_note_id_str} was NOT found in persistent_notifications list. The list remains unchanged.") + + self._save_persistent_history() + container.destroy() + self.containers = [c for c in self.containers if c != container] + self.rebuild_with_separators() + + def _add_historical_notification(self, note): + hist_notif = HistoricalNotification( + id=note.get("id"), + app_icon=note.get("app_icon"), + summary=note.get("summary"), + body=note.get("body"), + app_name=note.get("app_name"), + timestamp=note.get("timestamp"), + cached_image_path=note.get("cached_image_path"), + ) + + hist_box = NotificationBox(hist_notif, timeout_ms=0) + hist_box.uuid = hist_notif.id + hist_box.cached_image_path = hist_notif.cached_image_path + hist_box.set_is_history(True) + for child in hist_box.get_children(): + if child.get_name() == "notification-action-buttons": + hist_box.remove(child) + container = Box( + name="notification-container", + orientation="v", + h_align="fill", + h_expand=True, + ) + container.notification_box = hist_box + try: + arrival = datetime.fromisoformat(hist_notif.timestamp) + except Exception: + arrival = datetime.now() + container.arrival_time = arrival + + def compute_time_label(arrival_time): + return arrival_time.strftime("%H:%M") + + self.hist_time_label = Label( + name="notification-timestamp", + markup=compute_time_label(container.arrival_time), + h_align="start", + ellipsization="end", + ) + self.hist_notif_image_box = Box( + name="notification-image", + orientation="v", + children=[ + CustomImage( + pixbuf=load_scaled_pixbuf(hist_box, 48, 48) + ), + Box(v_expand=True), + ] + ) + self.hist_notif_summary_label = Label( + name="notification-summary", + markup=hist_notif.summary, + h_align="start", + + ellipsization="end", + + ) + + self.hist_notif_app_name_label = Label( + name="notification-app-name", + markup=f"{hist_notif.app_name}", + h_align="start", + + ellipsization="end", + + ) + + self.hist_notif_body_label = Label( + name="notification-body", + markup=hist_notif.body, + h_align="start", + + ellipsization="end", + line_wrap="word-char", + + ) if hist_notif.body else Box() + self.hist_notif_body_label.set_single_line_mode(True) if hist_notif.body else None + + self.hist_notif_summary_box = Box( + name="notification-summary-box", + orientation="h", + children=[ + self.hist_notif_summary_label, + Box(name="notif-sep", h_expand=False, v_expand=False, h_align="center", v_align="center"), + self.hist_notif_app_name_label, + Box(name="notif-sep", h_expand=False, v_expand=False, h_align="center", v_align="center"), + self.hist_time_label, + ], + ) + self.hist_notif_text_box = Box( + name="notification-text", + orientation="v", + v_align="center", + h_expand=True, + children=[ + self.hist_notif_summary_box, + self.hist_notif_body_label, + ], + ) + self.hist_notif_close_button = Button( + name="notif-close-button", + child=Label(name="notif-close-label", markup=icons.cancel), + on_clicked=lambda *_: self.delete_historical_notification(hist_notif.id, container), + ) + self.hist_notif_close_button_box = Box( + orientation="v", + children=[ + self.hist_notif_close_button, + Box(v_expand=True), + ], + ) + content_box = Box( + name="notification-box-hist", + spacing=8, + children=[ + self.hist_notif_image_box, + self.hist_notif_text_box, + self.hist_notif_close_button_box, + ], + ) + container.add(content_box) + self.containers.insert(0, container) + self.rebuild_with_separators() + self.update_no_notifications_label_visibility() + + def add_notification(self, notification_box): + app_name = notification_box.notification.app_name + if app_name in self.LIMITED_APPS_HISTORY: + self.clear_history_for_app(app_name) + + if len(self.containers) >= 50: + oldest_container = self.containers.pop() + if hasattr(oldest_container, "notification_box") and hasattr(oldest_container.notification_box, "cached_image_path") and oldest_container.notification_box.cached_image_path and os.path.exists(oldest_container.notification_box.cached_image_path): + try: + os.remove(oldest_container.notification_box.cached_image_path) + logger.info(f"Deleted cached image of oldest notification due to history limit: {oldest_container.notification_box.cached_image_path}") + except Exception as e: + logger.error(f"Error deleting cached image of oldest notification: {e}") + oldest_container.destroy() + + def on_container_destroy(container): + if hasattr(container, "_timestamp_timer_id") and container._timestamp_timer_id: + GLib.source_remove(container._timestamp_timer_id) + if hasattr(container, "notification_box"): + notif_box = container.notification_box + container.destroy() + self.containers.remove(container) + self.rebuild_with_separators() + self.update_no_notifications_label_visibility() + + container = Box( + name="notification-container", + orientation="v", + h_align="fill", + h_expand=True, + ) + container.arrival_time = datetime.now() + def compute_time_label(arrival_time): + return arrival_time.strftime("%H:%M") + self.current_time_label = Label(name="notification-timestamp", markup=compute_time_label(container.arrival_time)) + self.current_notif_image_box = Box( + name="notification-image", + orientation="v", + children=[ + CustomImage( + pixbuf=load_scaled_pixbuf(notification_box, 48, 48) + ), + Box(v_expand=True, v_align="fill"), + ] + ) + self.current_notif_summary_label = Label( + name="notification-summary", + markup=notification_box.notification.summary, + h_align="start", + ellipsization="end", + ) + self.current_notif_app_name_label = Label( + name="notification-app-name", + markup=f"{notification_box.notification.app_name}", + h_align="start", + ellipsization="end", + ) + self.current_notif_body_label = Label( + name="notification-body", + markup=notification_box.notification.body, + h_align="start", + ellipsization="end", + line_wrap="word-char", + ) if notification_box.notification.body else Box() + self.current_notif_body_label.set_single_line_mode(True) if notification_box.notification.body else None + self.current_notif_summary_box = Box( + name="notification-summary-box", + orientation="h", + children=[ + self.current_notif_summary_label, + Box(name="notif-sep", h_expand=False, v_expand=False, h_align="center", v_align="center"), + self.current_notif_app_name_label, + Box(name="notif-sep", h_expand=False, v_expand=False, h_align="center", v_align="center"), + self.current_time_label, + ], + ) + self.current_notif_text_box = Box( + name="notification-text", + orientation="v", + v_align="center", + h_expand=True, + children=[ + self.current_notif_summary_box, + self.current_notif_body_label, + ], + ) + self.current_notif_close_button = Button( + name="notif-close-button", + child=Label(name="notif-close-label", markup=icons.cancel), + on_clicked=lambda *_: on_container_destroy(container), + ) + self.current_notif_close_button_box = Box( + orientation="v", + children=[ + self.current_notif_close_button, + Box(v_expand=True), + ], + ) + content_box = Box( + name="notification-content", + spacing=8, + children=[ + self.current_notif_image_box, + self.current_notif_text_box, + self.current_notif_close_button_box, + ], + ) + container.notification_box = notification_box + hist_box = Box( + name="notification-box-hist", + orientation="v", + h_align="fill", + h_expand=True, + ) + hist_box.add(content_box) + content_box.get_children()[2].get_children()[0].connect( + "clicked", + lambda *_: on_container_destroy(container) + ) + container.add(hist_box) + self.containers.insert(0, container) + self.rebuild_with_separators() + self._append_persistent_notification(notification_box, container.arrival_time) + self.update_no_notifications_label_visibility() + + def _append_persistent_notification(self, notification_box, arrival_time): + note = { + "id": notification_box.uuid, + "app_icon": notification_box.notification.app_icon, + "summary": notification_box.notification.summary, + "body": notification_box.notification.body, + "app_name": notification_box.notification.app_name, + "timestamp": arrival_time.isoformat(), + "cached_image_path": notification_box.cached_image_path + } + self.persistent_notifications.insert(0, note) + self.persistent_notifications = self.persistent_notifications[:50] + self._save_persistent_history() + + def _cleanup_orphan_cached_images(self): + logger.debug("Starting orphan cached image cleanup.") + if not os.path.exists(PERSISTENT_DIR): + logger.debug("Cache directory does not exist, skipping cleanup.") + return + + cached_files = [f for f in os.listdir(PERSISTENT_DIR) if f.startswith("notification_") and f.endswith(".png")] + if not cached_files: + logger.debug("No cached image files found, skipping cleanup.") + return + + history_uuids = {note.get("id") for note in self.persistent_notifications if note.get("id")} + deleted_count = 0 + for cached_file in cached_files: + try: + uuid_from_filename = cached_file[len("notification_"):-len(".png")] + if uuid_from_filename not in history_uuids: + cache_file_path = os.path.join(PERSISTENT_DIR, cached_file) + os.remove(cache_file_path) + logger.info(f"Deleted orphan cached image: {cache_file_path}") + deleted_count += 1 + else: + logger.debug(f"Cached image {cached_file} found in history, keeping it.") + except Exception as e: + logger.error(f"Error processing cached file {cached_file} during cleanup: {e}") + + if deleted_count > 0: + logger.info(f"Orphan cached image cleanup finished. Deleted {deleted_count} images.") + else: + logger.info("Orphan cached image cleanup finished. No orphan images found.") + + def update_no_notifications_label_visibility(self): + has_notifications = bool(self.containers) + self.no_notifications_box.set_visible(not has_notifications) + self.notifications_list.set_visible(has_notifications) + + def clear_history_for_app(self, app_name): + """Clears all notifications in history for a specific app.""" + containers_to_remove = [] + persistent_notes_to_remove_ids = set() + for container in list(self.containers): + if hasattr(container, "notification_box") and container.notification_box.notification.app_name == app_name: + containers_to_remove.append(container) + persistent_notes_to_remove_ids.add(container.notification_box.uuid) + + for container in containers_to_remove: + if hasattr(container, "notification_box") and hasattr(container.notification_box, "cached_image_path") and container.notification_box.cached_image_path and os.path.exists(container.notification_box.cached_image_path): + try: + os.remove(container.notification_box.cached_image_path) + logger.info(f"Deleted cached image of replaced history notification: {container.notification_box.cached_image_path}") + except Exception as e: + logger.error(f"Error deleting cached image of replaced history notification: {e}") + self.containers.remove(container) + self.notifications_list.remove(container) + container.notification_box.destroy(from_history_delete=True) + container.destroy() + + self.persistent_notifications = [ + note for note in self.persistent_notifications + if note.get("id") not in persistent_notes_to_remove_ids + ] + self._save_persistent_history() + self.rebuild_with_separators() + self.update_no_notifications_label_visibility() + +class NotificationContainer(Box): + LIMITED_APPS = ["Spotify"] + + def __init__(self, notification_history_instance: NotificationHistory, revealer_transition_type: str = "slide-down"): + super().__init__(name="notification-container-main", orientation="v", spacing=4) + self.notification_history = notification_history_instance + + self._server = Notifications() + self._server.connect("notification-added", self.on_new_notification) + self._pending_removal = False + self._is_destroying = False + + self.stack = Gtk.Stack( + name="notification-stack", + transition_type=Gtk.StackTransitionType.SLIDE_LEFT_RIGHT, + transition_duration=200, + visible=True, + ) + self.navigation = Box( + name="notification-navigation", + spacing=4, + h_align="center" + ) + self.stack_box = Box( + name="notification-stack-box", + h_align="center", + h_expand=False, + children=[self.stack] + ) + self.prev_button = Button( + name="nav-button", + child=Label(name="nav-button-label", markup=icons.chevron_left), + on_clicked=self.show_previous, + ) + self.close_all_button = Button( + name="nav-button", + child=Label(name="nav-button-label", markup=icons.cancel), + on_clicked=self.close_all_notifications, + ) + self.close_all_button_label = self.close_all_button.get_child() + self.close_all_button_label.add_style_class("close") + self.next_button = Button( + name="nav-button", + child=Label(name="nav-button-label", markup=icons.chevron_right), + on_clicked=self.show_next, + ) + for button in [self.prev_button, self.close_all_button, self.next_button]: + button.connect("enter-notify-event", lambda *_: self.pause_and_reset_all_timeouts()) + button.connect("leave-notify-event", lambda *_: self.resume_all_timeouts()) + self.navigation.add(self.prev_button) + self.navigation.add(self.close_all_button) + self.navigation.add(self.next_button) + + self.navigation_revealer = Revealer( + transition_type="slide-down", + transition_duration=200, + child=self.navigation, + reveal_child=False, + ) + + self.notification_box_container = Box( + name="notification-box-internal-container", + orientation="v", + children=[self.stack_box, self.navigation_revealer] + ) + + self.main_revealer = Revealer( + name="notification-main-revealer", + transition_type=revealer_transition_type, + transition_duration=250, + child_revealed=False, + child=self.notification_box_container, + ) + + self.add(self.main_revealer) + + self.notifications = [] + self.current_index = 0 + self.update_navigation_buttons() + self._destroyed_notifications = set() + + def on_new_notification(self, fabric_notif, id): + notification_history_instance = self.notification_history + if notification_history_instance.do_not_disturb_enabled: + logger.info("Do Not Disturb mode enabled: adding notification directly to history.") + notification = fabric_notif.get_notification_from_id(id) + new_box = NotificationBox(notification) + if notification.image_pixbuf: + cache_notification_pixbuf(new_box) + notification_history_instance.add_notification(new_box) + return + + notification = fabric_notif.get_notification_from_id(id) + new_box = NotificationBox(notification) + new_box.set_container(self) + notification.connect("closed", self.on_notification_closed) + + app_name = notification.app_name + if app_name in self.LIMITED_APPS: + notification_history_instance.clear_history_for_app(app_name) + + existing_notification_index = -1 + for index, existing_box in enumerate(self.notifications): + if existing_box.notification.app_name == app_name: + existing_notification_index = index + break + + if existing_notification_index != -1: + old_notification_box = self.notifications.pop(existing_notification_index) + self.stack.remove(old_notification_box) + old_notification_box.destroy() + + self.stack.add_named(new_box, str(id)) + self.notifications.append(new_box) + self.current_index = len(self.notifications) - 1 + self.stack.set_visible_child(new_box) + else: + while len(self.notifications) >= 5: + oldest_notification = self.notifications[0] + notification_history_instance.add_notification(oldest_notification) + self.stack.remove(oldest_notification) + self.notifications.pop(0) + if self.current_index > 0: + self.current_index -= 1 + self.stack.add_named(new_box, str(id)) + self.notifications.append(new_box) + self.current_index = len(self.notifications) - 1 + self.stack.set_visible_child(new_box) + else: + while len(self.notifications) >= 5: + oldest_notification = self.notifications[0] + notification_history_instance.add_notification(oldest_notification) + self.stack.remove(oldest_notification) + self.notifications.pop(0) + if self.current_index > 0: + self.current_index -= 1 + self.stack.add_named(new_box, str(id)) + self.notifications.append(new_box) + self.current_index = len(self.notifications) - 1 + self.stack.set_visible_child(new_box) + + for notification_box in self.notifications: + notification_box.start_timeout() + self.main_revealer.show_all() + self.main_revealer.set_reveal_child(True) + self.update_navigation_buttons() + + def show_previous(self, *args): + if self.current_index > 0: + self.current_index -= 1 + self.stack.set_visible_child(self.notifications[self.current_index]) + self.update_navigation_buttons() + + def show_next(self, *args): + if self.current_index < len(self.notifications) - 1: + self.current_index += 1 + self.stack.set_visible_child(self.notifications[self.current_index]) + self.update_navigation_buttons() + + def update_navigation_buttons(self): + self.prev_button.set_sensitive(self.current_index > 0) + self.next_button.set_sensitive(self.current_index < len(self.notifications) - 1) + should_reveal = len(self.notifications) > 1 + self.navigation_revealer.set_reveal_child(should_reveal) + + def on_notification_closed(self, notification, reason): + if self._is_destroying: + return + if notification.id in self._destroyed_notifications: + return + self._destroyed_notifications.add(notification.id) + try: + logger.info(f"Notification {notification.id} closing with reason: {reason}") + notif_to_remove = None + for i, notif_box in enumerate(self.notifications): + if notif_box.notification.id == notification.id: + notif_to_remove = (i, notif_box) + break + if not notif_to_remove: + return + i, notif_box = notif_to_remove + reason_str = str(reason) + + notification_history_instance = self.notification_history + + if reason_str == "NotificationCloseReason.DISMISSED_BY_USER": + logger.info(f"Cleaning up resources for dismissed notification {notification.id}") + notif_box.destroy() + elif (reason_str == "NotificationCloseReason.EXPIRED" or + reason_str == "NotificationCloseReason.CLOSED" or + reason_str == "NotificationCloseReason.UNDEFINED"): + logger.info(f"Adding notification {notification.id} to history (reason: {reason_str})") + notif_box.set_is_history(True) + notification_history_instance.add_notification(notif_box) + notif_box.stop_timeout() + else: + logger.warning(f"Unknown close reason: {reason_str} for notification {notification.id}. Defaulting to destroy.") + notif_box.destroy() + + if len(self.notifications) == 1: + self._is_destroying = True + self.main_revealer.set_reveal_child(False) + GLib.timeout_add( + self.main_revealer.get_transition_duration(), + self._destroy_container + ) + return + + new_index = i + if i == self.current_index: + new_index = max(0, i - 1) + elif i < self.current_index: + new_index = self.current_index - 1 + + if notif_box.get_parent() == self.stack: + self.stack.remove(notif_box) + self.notifications.pop(i) + + if new_index >= len(self.notifications) and len(self.notifications) > 0: + new_index = len(self.notifications) - 1 + + self.current_index = new_index + + if len(self.notifications) > 0: + self.stack.set_visible_child(self.notifications[self.current_index]) + + self.update_navigation_buttons() + except Exception as e: + logger.error(f"Error closing notification: {e}") + + def _destroy_container(self): + try: + self.notifications.clear() + self._destroyed_notifications.clear() + for child in self.stack.get_children(): + self.stack.remove(child) + child.destroy() + self.current_index = 0 + except Exception as e: + logger.error(f"Error cleaning up the container: {e}") + finally: + self._is_destroying = False + return False + + def pause_and_reset_all_timeouts(self): + if self._is_destroying: + return + for notification in self.notifications[:]: + try: + if not notification._destroyed and notification.get_parent(): + notification.stop_timeout() + except Exception as e: + logger.error(f"Error pausing timeout: {e}") + + def resume_all_timeouts(self): + if self._is_destroying: + return + for notification in self.notifications[:]: + try: + if not notification._destroyed and notification.get_parent(): + notification.start_timeout() + except Exception as e: + logger.error(f"Error resuming timeout: {e}") + + def close_all_notifications(self, *args): + notifications_to_close = self.notifications.copy() + for notification_box in notifications_to_close: + notification_box.notification.close("dismissed-by-user") + +class NotificationPopup(Window): + def __init__(self, **kwargs): + y_pos = data.NOTIF_POS.lower() + x_pos = "right" + + if data.BAR_POSITION in ["Top", "Bottom"] and data.PANEL_POSITION == "End" or x_pos == data.BAR_POSITION.lower(): + x_pos = "left" + + super().__init__( + name="notification-popup", + anchor=f"{x_pos} {y_pos}", + layer="top", + keyboard_mode="none", + exclusivity="none", + visible=True, + all_visible=True, + ) + + self.widgets = kwargs.get("widgets", None) + + self.notification_history = self.widgets.notification_history if self.widgets else NotificationHistory() + self.notification_container = NotificationContainer( + notification_history_instance=self.notification_history, + revealer_transition_type="slide-down" if y_pos == "top" else "slide-up" + ) + + self.show_box = Box() + self.show_box.set_size_request(1, 1) + + self.add(Box( + name="notification-popup-box", + orientation="v", + children=[self.notification_container, self.show_box] + )) diff --git a/modules/dock/components/notifications.py b/modules/dock/components/notifications.py index 146ac52..e82b57a 100644 --- a/modules/dock/components/notifications.py +++ b/modules/dock/components/notifications.py @@ -1,70 +1,89 @@ import json import locale import os -import uuid from datetime import datetime, timedelta +from fabric.utils.helpers import get_relative_path from fabric.widgets.box import Box from fabric.widgets.button import Button from fabric.widgets.centerbox import CenterBox from fabric.widgets.label import Label from fabric.widgets.scrolledwindow import ScrolledWindow -from gi.repository import GdkPixbuf, GLib, Gtk -from loguru import logger +from gi.repository import GLib, Gtk, Gdk import config.data as data import utils.icons as icons -from modules.notification_popup import NotificationBox, cache_notification_pixbuf, load_scaled_pixbuf +from modules.notification_popup import NotificationBox, load_scaled_pixbuf, get_shared_notification_history from utils.custom_image import CustomImage from utils.wayland import WaylandWindow as Window PERSISTENT_DIR = f"/tmp/{data.APP_NAME}/notifications" PERSISTENT_HISTORY_FILE = os.path.join(PERSISTENT_DIR, "notification_history.json") +CONFIG_FILE = get_relative_path("../../../config/assets/config.json") class NotificationIndicator(Button): def __init__(self, **kwargs): super().__init__(name="button-bar-notifications", **kwargs) - # Create the notification icon + self.notification_icon = Label( - name="notification-icon", - markup=icons.notifications + name="notification-icon", markup=icons.notifications ) - # Add the icon directly self.add(self.notification_icon) - # Connect click handler self.connect("clicked", self.on_clicked) - # Set default tooltip self.set_tooltip_text("Notifications") - # Popup window for showing notifications self.popup_window = None + + self._connect_to_dnd_state() + def on_clicked(self, _button): - """Handle click to show/hide notification popup.""" if self.popup_window and self.popup_window.get_visible(): self.popup_window.set_visible(False) else: self.show_notifications_popup() def show_notifications_popup(self): - """Show popup window with notification history.""" if self.popup_window: self.popup_window.destroy() - print(f"[NotificationIndicator] Opening notification history popup") - - # Create popup window using the notification history directly self.popup_window = NotificationHistoryWindow() self.popup_window.show_all() + def _connect_to_dnd_state(self): + try: + + self.notification_history = get_shared_notification_history() + + + self.notification_history.connect('dnd-state-changed', self._on_dnd_state_changed) + + + self._update_icon_for_dnd_state(self.notification_history.do_not_disturb_enabled) + except Exception as e: + print(f"Could not connect to DND state: {e}") + + def _on_dnd_state_changed(self, notification_history, dnd_enabled): + self._update_icon_for_dnd_state(dnd_enabled) + + def _update_icon_for_dnd_state(self, dnd_enabled): + if dnd_enabled: + self.notification_icon.set_markup(icons.notifications_off) + self.set_tooltip_text("Do Not Disturb enabled") + else: + self.notification_icon.set_markup(icons.notifications) + self.set_tooltip_text("Notifications") + class HistoricalNotification(object): - def __init__(self, id, app_icon, summary, body, app_name, timestamp, cached_image_path=None): + def __init__( + self, id, app_icon, summary, body, app_name, timestamp, cached_image_path=None + ): self.id = id self.app_icon = app_icon self.summary = summary @@ -79,11 +98,7 @@ class HistoricalNotification(object): class NotificationHistory(Box): def __init__(self, **kwargs): - super().__init__( - name="notification-history", - orientation="v", - **kwargs - ) + super().__init__(name="notification-history", orientation="v", **kwargs) self.containers = [] self.header_label = Label( @@ -148,7 +163,10 @@ class NotificationHistory(Box): propagate_width=False, propagate_height=False, ) - self.scrolled_window_viewport_box = Box(orientation="v", children=[self.notifications_list, self.no_notifications_box]) + self.scrolled_window_viewport_box = Box( + orientation="v", + children=[self.notifications_list, self.no_notifications_box], + ) self.scrolled_window.add_with_viewport(self.scrolled_window_viewport_box) self.persistent_notifications = [] self.add(self.history_header) @@ -157,13 +175,72 @@ class NotificationHistory(Box): self._cleanup_orphan_cached_images() self.schedule_midnight_update() + + self._connect_to_shared_history() + + + self._load_and_sync_dnd_state() + self.LIMITED_APPS_HISTORY = ["Spotify"] + def _connect_to_shared_history(self): + try: + shared_history = get_shared_notification_history() + shared_history.connect('notification-added', self._on_shared_notification_added) + except Exception as e: + print(f"Could not connect to shared notification history: {e}") + + def _on_shared_notification_added(self, shared_history): + GLib.idle_add(self._refresh_history) + + def _refresh_history(self): + try: + + for child in self.notifications_list.get_children()[:]: + self.notifications_list.remove(child) + if hasattr(child, 'destroy'): + child.destroy() + + self.containers.clear() + + + if os.path.exists(PERSISTENT_HISTORY_FILE): + with open(PERSISTENT_HISTORY_FILE, "r") as f: + self.persistent_notifications = json.load(f) + + + for note in reversed(self.persistent_notifications[-50:]): + self._add_historical_notification(note) + + self.update_no_notifications_label_visibility() + except Exception as e: + print(f"Error refreshing notification history: {e}") + + def _load_and_sync_dnd_state(self): + try: + + dnd_enabled = False + if os.path.exists(CONFIG_FILE): + with open(CONFIG_FILE, "r") as f: + config_data = json.load(f) + dnd_enabled = config_data.get("dnd_enabled", False) + + + shared_history = get_shared_notification_history() + shared_history.do_not_disturb_enabled = dnd_enabled + + + self.header_switch.set_active(dnd_enabled) + self.do_not_disturb_enabled = dnd_enabled + + except Exception as e: + print(f"Could not sync DND state: {e}") + def get_ordinal(self, n): if 11 <= (n % 100) <= 13: - return 'th' + return "th" else: - return {1: 'st', 2: 'nd', 3: 'rd'}.get(n % 10, 'th') + return {1: "st", 2: "nd", 3: "rd"}.get(n % 10, "th") def get_date_header(self, dt): now = datetime.now() @@ -176,9 +253,9 @@ class NotificationHistory(Box): else: original_locale = locale.getlocale(locale.LC_TIME) try: - locale.setlocale(locale.LC_TIME, ('en_US', 'UTF-8')) + locale.setlocale(locale.LC_TIME, ("en_US", "UTF-8")) except locale.Error: - locale.setlocale(locale.LC_TIME, 'C') + locale.setlocale(locale.LC_TIME, "C") try: day = dt.day ordinal = self.get_ordinal(day) @@ -193,7 +270,9 @@ class NotificationHistory(Box): def schedule_midnight_update(self): now = datetime.now() - next_midnight = datetime.combine(now.date() + timedelta(days=1), datetime.min.time()) + next_midnight = datetime.combine( + now.date() + timedelta(days=1), datetime.min.time() + ) delta_seconds = (next_midnight - now).total_seconds() GLib.timeout_add_seconds(int(delta_seconds), self.on_midnight) @@ -212,7 +291,7 @@ class NotificationHistory(Box): h_align="center", h_expand=True, ) - ] + ], ) def rebuild_with_separators(self): @@ -225,7 +304,9 @@ class NotificationHistory(Box): current_date_header = None last_date_header = None - for container in sorted(self.containers, key=lambda x: x.arrival_time, reverse=True): + for container in sorted( + self.containers, key=lambda x: x.arrival_time, reverse=True + ): arrival_time = container.arrival_time date_header = self.get_date_header(arrival_time) if date_header != current_date_header: @@ -245,56 +326,58 @@ class NotificationHistory(Box): def on_do_not_disturb_changed(self, switch, pspec): self.do_not_disturb_enabled = switch.get_active() - logger.info(f"Do Not Disturb mode {'enabled' if self.do_not_disturb_enabled else 'disabled'}") + + + + try: + shared_history = get_shared_notification_history() + shared_history.set_do_not_disturb_enabled(self.do_not_disturb_enabled) + except Exception as e: + print(f"Could not update shared DND state: {e}") def clear_history(self, *args): for child in self.notifications_list.get_children()[:]: container = child - notif_box = container.notification_box if hasattr(container, "notification_box") else None + notif_box = ( + container.notification_box + if hasattr(container, "notification_box") + else None + ) if notif_box: notif_box.destroy(from_history_delete=True) self.notifications_list.remove(child) child.destroy() - # Clear notifications from the fabric service - try: - # Clear all notifications from the fabric service - # Note: fabric Notifications doesn't have clear_all_notifications method - # so we just clear our local history - logger.info("Notification history cleared.") - except Exception as e: - logger.error(f"Error clearing notification history: {e}") - # Also clear persistent file if it exists + if os.path.exists(PERSISTENT_HISTORY_FILE): try: os.remove(PERSISTENT_HISTORY_FILE) - logger.info("Notification history cleared and persistent file deleted.") + except Exception as e: - logger.error(f"Error deleting persistent history file: {e}") + print(f"Error deleting persistent history file: {e}") self.persistent_notifications = [] self.containers = [] self.rebuild_with_separators() def _load_persistent_history(self): - """Load notifications from persistent file.""" if not os.path.exists(PERSISTENT_DIR): os.makedirs(PERSISTENT_DIR, exist_ok=True) - # Load from persistent file + if os.path.exists(PERSISTENT_HISTORY_FILE): try: with open(PERSISTENT_HISTORY_FILE, "r") as f: self.persistent_notifications = json.load(f) - logger.info(f"Loading {len(self.persistent_notifications)} notifications from persistent history") - # Add notifications to display (in reverse order to show newest first) - for note in reversed(self.persistent_notifications[-50:]): # Last 50 notifications + + + for note in reversed(self.persistent_notifications[-50:]): self._add_historical_notification(note) except Exception as e: - logger.error(f"Error loading persistent history: {e}") + print(f"Error loading persistent history: {e}") self.persistent_notifications = [] else: self.persistent_notifications = [] @@ -306,7 +389,7 @@ class NotificationHistory(Box): with open(PERSISTENT_HISTORY_FILE, "w") as f: json.dump(self.persistent_notifications, f) except Exception as e: - logger.error(f"Error saving persistent history: {e}") + print(f"Error saving persistent history: {e}") def update_no_notifications_label_visibility(self): has_notifications = bool(self.containers) @@ -314,35 +397,39 @@ class NotificationHistory(Box): self.notifications_list.set_visible(has_notifications) def _cleanup_orphan_cached_images(self): - logger.debug("Starting orphan cached image cleanup.") + if not os.path.exists(PERSISTENT_DIR): - logger.debug("Cache directory does not exist, skipping cleanup.") + return - cached_files = [f for f in os.listdir(PERSISTENT_DIR) if f.startswith("notification_") and f.endswith(".png")] + cached_files = [ + f + for f in os.listdir(PERSISTENT_DIR) + if f.startswith("notification_") and f.endswith(".png") + ] if not cached_files: - logger.debug("No cached image files found, skipping cleanup.") + return - history_uuids = {note.get("id") for note in self.persistent_notifications if note.get("id")} + history_uuids = { + note.get("id") for note in self.persistent_notifications if note.get("id") + } deleted_count = 0 for cached_file in cached_files: try: - uuid_from_filename = cached_file[len("notification_"):-len(".png")] + uuid_from_filename = cached_file[len("notification_") : -len(".png")] if uuid_from_filename not in history_uuids: cache_file_path = os.path.join(PERSISTENT_DIR, cached_file) os.remove(cache_file_path) - logger.info(f"Deleted orphan cached image: {cache_file_path}") - deleted_count += 1 - else: - logger.debug(f"Cached image {cached_file} found in history, keeping it.") - except Exception as e: - logger.error(f"Error processing cached file {cached_file} during cleanup: {e}") - if deleted_count > 0: - logger.info(f"Orphan cached image cleanup finished. Deleted {deleted_count} images.") - else: - logger.info("Orphan cached image cleanup finished. No orphan images found.") + deleted_count += 1 + + except Exception as e: + print( + f"Error processing cached file {cached_file} during cleanup: {e}" + ) + + def _add_historical_notification(self, note): hist_notif = HistoricalNotification( @@ -388,11 +475,9 @@ class NotificationHistory(Box): name="notification-image", orientation="v", children=[ - CustomImage( - pixbuf=load_scaled_pixbuf(hist_box, 48, 48) - ), + CustomImage(pixbuf=load_scaled_pixbuf(hist_box, 48, 48)), Box(v_expand=True), - ] + ], ) self.hist_notif_summary_label = Label( name="notification-summary", @@ -408,23 +493,41 @@ class NotificationHistory(Box): ellipsization="end", ) - self.hist_notif_body_label = Label( - name="notification-body", - markup=hist_notif.body, - h_align="start", - ellipsization="end", - line_wrap="word-char", - ) if hist_notif.body else Box() - self.hist_notif_body_label.set_single_line_mode(True) if hist_notif.body else None + self.hist_notif_body_label = ( + Label( + name="notification-body", + markup=hist_notif.body, + h_align="start", + ellipsization="end", + line_wrap="word-char", + ) + if hist_notif.body + else Box() + ) + self.hist_notif_body_label.set_single_line_mode( + True + ) if hist_notif.body else None self.hist_notif_summary_box = Box( name="notification-summary-box", orientation="h", children=[ self.hist_notif_summary_label, - Box(name="notif-sep", h_expand=False, v_expand=False, h_align="center", v_align="center"), + Box( + name="notif-sep", + h_expand=False, + v_expand=False, + h_align="center", + v_align="center", + ), self.hist_notif_app_name_label, - Box(name="notif-sep", h_expand=False, v_expand=False, h_align="center", v_align="center"), + Box( + name="notif-sep", + h_expand=False, + v_expand=False, + h_align="center", + v_align="center", + ), self.hist_time_label, ], ) @@ -441,7 +544,9 @@ class NotificationHistory(Box): self.hist_notif_close_button = Button( name="notif-close-button", child=Label(name="notif-close-label", markup=icons.cancel), - on_clicked=lambda *_: self.delete_historical_notification(hist_notif.id, container), + on_clicked=lambda *_: self.delete_historical_notification( + hist_notif.id, container + ), ) self.hist_notif_close_button_box = Box( orientation="v", @@ -482,9 +587,8 @@ class NotificationHistory(Box): if removed_from_list: self.persistent_notifications = new_persistent_notifications - logger.info(f"Notification with ID {target_note_id_str} was marked for removal from persistent_notifications list.") - else: - logger.warning(f"Notification with ID {target_note_id_str} was NOT found in persistent_notifications list. The list remains unchanged.") + + self._save_persistent_history() container.destroy() @@ -494,10 +598,10 @@ class NotificationHistory(Box): class NotificationHistoryWindow(Window): def __init__(self): - # Create the notification history widget directly + self.notification_history = NotificationHistory() - # Create scrolled window with proper sizing + scrolled = ScrolledWindow( name="notifications-scrolled", child=self.notification_history, @@ -508,7 +612,7 @@ class NotificationHistoryWindow(Window): h_expand=True, v_expand=True, propagate_width=False, - propagate_height=False + propagate_height=False, ) main_box = Box( @@ -517,20 +621,17 @@ class NotificationHistoryWindow(Window): spacing=8, h_expand=True, v_expand=True, - children=[scrolled] + children=[scrolled], ) - # Determine popup position based on dock position + dock_position = data.DOCK_POSITION popup_anchor = self._get_popup_anchor(dock_position) - print(f"[NotificationPopup] Dock position: {dock_position}, Popup anchor: {popup_anchor}") - # Get margin based on dock position margin = self._get_popup_margin(dock_position) super().__init__( - name="notifications-popup-window", anchor=popup_anchor, margin=margin, child=main_box, @@ -538,89 +639,75 @@ class NotificationHistoryWindow(Window): exclusive=False, keyboard_mode="on-demand", visible=True, - all_visible=True + all_visible=True, ) - # Auto-hide when clicking outside or pressing escape + self.connect("button-press-event", self.on_button_press) self.connect("key-press-event", self.on_key_press) - # Make sure the window can receive key events + self.set_can_focus(True) self.grab_focus() - # Set tooltip with keyboard shortcuts - self.set_tooltip_text("Keyboard shortcuts:\n• Escape: Close popup\n• Ctrl+D: Toggle DND\n• Ctrl+A: Clear all notifications") + + self.set_tooltip_text( + "Keyboard shortcuts:\n• Escape: Close popup\n• Ctrl+D: Toggle DND\n• Ctrl+A: Clear all notifications" + ) def _get_popup_anchor(self, dock_position): - """Determine popup anchor position based on dock position.""" - # Position popup on the opposite side of the dock for better UX anchor_map = { - "Top": "top", # Dock at top -> popup at top right - "Bottom": "bottom", # Dock at bottom -> popup at bottom right - "Left": "left", # Dock at left -> popup at top left - "Right": "right" # Dock at right -> popup at top right + "Top": "top", # Dock at top -> popup at top right + "Bottom": "bottom", # Dock at bottom -> popup at bottom right + "Left": "left", # Dock at left -> popup at top left + "Right": "right", # Dock at right -> popup at top right } - return anchor_map.get(dock_position, "bottom") # Default fallback + return anchor_map.get(dock_position, "bottom") def _get_popup_margin(self, dock_position): - """Get appropriate margin based on dock position.""" - # Add margin to avoid overlapping with dock margin_map = { - "Top": "60px 10px 10px 10px", # Top dock -> margin from top - "Bottom": "10px 10px 60px 10px", # Bottom dock -> margin from bottom - "Left": "10px 10px 10px 60px", # Left dock -> margin from left - "Right": "10px 60px 10px 10px" # Right dock -> margin from right + "Top": "60px 10px 10px 10px", + "Bottom": "10px 10px 60px 10px", + "Left": "10px 10px 10px 60px", + "Right": "10px 60px 10px 10px", } - return margin_map.get(dock_position, "10px 10px 10px 10px") # Default fallback + return margin_map.get(dock_position, "10px 10px 10px 10px") def refresh_popup(self): - """Refresh the popup content by updating the notification history.""" - # The notification history will automatically update itself - # when notifications are added/removed from the service pass def on_key_press(self, _widget, event): - """Handle key press events.""" - from gi.repository import Gdk - # Check if Escape key was pressed if event.keyval == Gdk.KEY_Escape: - print(f"[NotificationPopup] Escape key pressed, closing popup") - self.set_visible(False) - return True # Event handled - # Check for Ctrl+D to toggle DND + self.set_visible(False) + return True + + elif event.state & Gdk.ModifierType.CONTROL_MASK and event.keyval == Gdk.KEY_d: - print(f"[NotificationPopup] Ctrl+D pressed, toggling DND") - # Toggle DND via the notification history + current_dnd = self.notification_history.do_not_disturb_enabled self.notification_history.header_switch.set_active(not current_dnd) - return True # Event handled + return True + - # Check for Ctrl+A to clear all notifications elif event.state & Gdk.ModifierType.CONTROL_MASK and event.keyval == Gdk.KEY_a: - print(f"[NotificationPopup] Ctrl+A pressed, clearing all notifications") + self.notification_history.clear_history() - return True # Event handled + return True - return False # Let other handlers process the event - - def on_button_press(self, _widget, _event): - """Hide popup when clicking outside.""" - # This is a simple implementation - in a real scenario you'd want - # more sophisticated outside-click detection return False + def on_button_press(self, _widget, _event): + return False class Notifications(Box): def __init__(self, **kwargs): super().__init__( - name="notifications", orientation="h" if not data.VERTICAL else "v", spacing=4, children=[NotificationIndicator()], - **kwargs + **kwargs, ) self.show_all() diff --git a/modules/launcher/plugin_manager.py b/modules/launcher/plugin_manager.py index 56c89ff..1fe2ebb 100644 --- a/modules/launcher/plugin_manager.py +++ b/modules/launcher/plugin_manager.py @@ -93,7 +93,6 @@ class PluginManager: "network", "bluetooth", "otp", - "notifications", "password", "bookmarks", "process", diff --git a/modules/launcher/plugins/notifications.py b/modules/launcher/plugins/archive/notifications.py similarity index 100% rename from modules/launcher/plugins/notifications.py rename to modules/launcher/plugins/archive/notifications.py diff --git a/modules/launcher/plugins/process.py b/modules/launcher/plugins/process.py index f631249..d94bfbe 100644 --- a/modules/launcher/plugins/process.py +++ b/modules/launcher/plugins/process.py @@ -291,9 +291,7 @@ class ProcessPlugin(PluginBase): ) mem_indicator = "💾" if memory_percent > 80 else "" - subtitle = f"CPU: {cpu_percent:.1f}%{cpu_indicator} | Memory: { - memory_mb:.1f - }MB ({memory_percent:.1f}%){mem_indicator}" + subtitle = f"CPU: {cpu_percent:.1f}%{cpu_indicator} | Memory: {memory_mb:.1f}MB ({memory_percent:.1f}%){mem_indicator}" # Choose icon based on CPU usage if cpu_percent > 50: @@ -344,11 +342,6 @@ class ProcessPlugin(PluginBase): if self._kill_process(pid, silent=True): killed_count += 1 - print( - f"✓ Terminated {killed_count}/{len(child_pids)} processes for '{ - app_name - }'" - ) except Exception as e: print(f"✗ Error killing process group: {e}") @@ -365,10 +358,6 @@ class ProcessPlugin(PluginBase): # Wait a bit for graceful termination try: proc.wait(timeout=3) - if not silent: - print( - f"✓ Successfully terminated process '{proc_name}' (PID: {pid})" - ) return True except psutil.TimeoutExpired: # Force kill if graceful termination failed @@ -497,9 +486,7 @@ class ProcessPlugin(PluginBase): "🔥" if cpu_percent > 80 else "⚡" if cpu_percent > 50 else "" ) mem_indicator = "💾" if memory_percent > 80 else "" - new_subtitle = f"CPU: {cpu_percent:.1f}%{cpu_indicator} | Memory: { - memory_mb:.1f - }MB ({memory_percent:.1f}%){mem_indicator}" + new_subtitle = f"CPU: {cpu_percent:.1f}%{cpu_indicator} | Memory: {memory_mb:.1f}MB ({memory_percent:.1f}%){mem_indicator}" # Find and update the subtitle label widget self._find_and_update_subtitle_label(result_item, new_subtitle) diff --git a/modules/notification_popup.py b/modules/notification_popup.py index 02c3e8d..b1913ca 100644 --- a/modules/notification_popup.py +++ b/modules/notification_popup.py @@ -1,18 +1,17 @@ import json -import locale import os import uuid -from datetime import datetime, timedelta +from datetime import datetime from fabric.notifications.service import Notification, NotificationAction, Notifications +from fabric.utils.helpers import get_relative_path from fabric.widgets.box import Box from fabric.widgets.button import Button from fabric.widgets.centerbox import CenterBox from fabric.widgets.image import Image from fabric.widgets.label import Label from fabric.widgets.revealer import Revealer -from fabric.widgets.scrolledwindow import ScrolledWindow -from gi.repository import GdkPixbuf, GLib, Gtk +from gi.repository import GdkPixbuf, GLib, GObject, Gtk from loguru import logger import config.data as data @@ -22,31 +21,33 @@ from utils.wayland import WaylandWindow as Window PERSISTENT_DIR = f"/tmp/{data.APP_NAME}/notifications" PERSISTENT_HISTORY_FILE = os.path.join(PERSISTENT_DIR, "notification_history.json") +CONFIG_FILE = get_relative_path("../config/assets/config.json") + +# Global shared notification history instance for DND state +_shared_notification_history = None + + +def get_shared_notification_history(): + global _shared_notification_history + if _shared_notification_history is None: + _shared_notification_history = NotificationHistory() + return _shared_notification_history def cache_notification_pixbuf(notification_box): - """ - Saves a scaled pixbuf (48x48) in the cache directory and returns the cache file path. - """ notification = notification_box.notification if notification.image_pixbuf: os.makedirs(PERSISTENT_DIR, exist_ok=True) cache_file = os.path.join( PERSISTENT_DIR, f"notification_{notification_box.uuid}.png" ) - logger.debug( - f"Caching image for notification {notification.id} to: {cache_file}" - ) + try: scaled = notification.image_pixbuf.scale_simple( 48, 48, GdkPixbuf.InterpType.BILINEAR ) scaled.savev(cache_file, "png", [], []) - logger.info( - f"Successfully cached image for notification {notification.id} to: { - cache_file - }" - ) + return cache_file except Exception as e: logger.error(f"Error caching image for notification {notification.id}: {e}") @@ -57,14 +58,8 @@ def cache_notification_pixbuf(notification_box): def load_scaled_pixbuf(notification_box, width, height): - """ - Loads and scales a pixbuf for a notification_box, prioritizing cached images. - """ notification = notification_box.notification if not hasattr(notification_box, "notification") or notification is None: - logger.error( - "load_scaled_pixbuf: notification_box.notification is None or not set!" - ) return None pixbuf = None @@ -84,11 +79,7 @@ def load_scaled_pixbuf(notification_box, width, height): pixbuf = pixbuf.scale_simple( width, height, GdkPixbuf.InterpType.BILINEAR ) - logger.info( - f"Successfully loaded cached image from: { - notification_box.cached_image_path - } for notification {notification.id}" - ) + return pixbuf except Exception as e: logger.error( @@ -103,28 +94,15 @@ def load_scaled_pixbuf(notification_box, width, height): ) if notification.image_pixbuf: - logger.debug( - f"Loading image directly from notification.image_pixbuf for notification { - notification.id - }" - ) pixbuf = notification.image_pixbuf.scale_simple( width, height, GdkPixbuf.InterpType.BILINEAR ) return pixbuf - logger.debug( - f"No image_pixbuf or cached image found, trying app icon for notification { - notification.id - }" - ) return get_app_icon_pixbuf(notification.app_icon, width, height) def get_app_icon_pixbuf(icon_path, width, height): - """ - Loads and scales a pixbuf from an app icon path. - """ if not icon_path: return None if icon_path.startswith("file://"): @@ -207,11 +185,7 @@ class NotificationBox(Box): cache_path = cache_notification_pixbuf(self) if cache_path: self.cached_image_path = cache_path - logger.debug( - f"NotificationBox {self.uuid}: Cached image path set to: { - self.cached_image_path - }" - ) + else: logger.warning( f"NotificationBox { @@ -232,9 +206,6 @@ class NotificationBox(Box): self._destroyed = False self._is_history = False - logger.debug( - f"NotificationBox {self.uuid} created for notification {notification.id}" - ) def set_is_history(self, is_history): self._is_history = is_history @@ -420,13 +391,6 @@ class NotificationBox(Box): return False def destroy(self, from_history_delete=False): - logger.debug( - f"NotificationBox destroy called for notification: { - self.notification.id - }, from_history_delete: {from_history_delete}, is_history: { - self._is_history - }" - ) if ( hasattr(self, "cached_image_path") and self.cached_image_path @@ -454,14 +418,25 @@ class NotificationBox(Box): class NotificationHistory(Box): + __gsignals__ = { + "dnd-state-changed": (GObject.SignalFlags.RUN_FIRST, None, (bool,)), + "notification-added": (GObject.SignalFlags.RUN_FIRST, None, ()), + } + def __init__(self, **kwargs): super().__init__(name="notification-history", orientation="v", **kwargs) self.do_not_disturb_enabled = False self.persistent_notifications = [] self._load_persistent_history() + self._load_dnd_state() + + def set_do_not_disturb_enabled(self, enabled): + if self.do_not_disturb_enabled != enabled: + self.do_not_disturb_enabled = enabled + self._save_dnd_state() + self.emit("dnd-state-changed", enabled) def _load_persistent_history(self): - """Load notifications from persistent file.""" if not os.path.exists(PERSISTENT_DIR): os.makedirs(PERSISTENT_DIR, exist_ok=True) @@ -480,8 +455,33 @@ class NotificationHistory(Box): else: self.persistent_notifications = [] + def _load_dnd_state(self): + try: + if os.path.exists(CONFIG_FILE): + with open(CONFIG_FILE, "r") as f: + config_data = json.load(f) + self.do_not_disturb_enabled = config_data.get("dnd_enabled", False) + logger.info(f"Loaded DND state: {self.do_not_disturb_enabled}") + except Exception as e: + logger.error(f"Error loading DND state: {e}") + self.do_not_disturb_enabled = False + + def _save_dnd_state(self): + try: + config_data = {} + if os.path.exists(CONFIG_FILE): + with open(CONFIG_FILE, "r") as f: + config_data = json.load(f) + + config_data["dnd_enabled"] = self.do_not_disturb_enabled + + with open(CONFIG_FILE, "w") as f: + json.dump(config_data, f, indent=4) + logger.debug(f"Saved DND state: {self.do_not_disturb_enabled}") + except Exception as e: + logger.error(f"Error saving DND state: {e}") + def _save_persistent_history(self): - """Save notifications to persistent file.""" try: with open(PERSISTENT_HISTORY_FILE, "w") as f: json.dump(self.persistent_notifications, f, indent=2) @@ -494,11 +494,9 @@ class NotificationHistory(Box): logger.error(f"Error saving persistent history: {e}") def add_notification(self, notification_box): - """Add a notification to the persistent history.""" try: notification = notification_box.notification - # Convert to persistent format hist_data = { "id": notification_box.uuid, "app_icon": getattr( @@ -512,28 +510,18 @@ class NotificationHistory(Box): notification_box, "cached_image_path", None ), } - - # Add to persistent notifications (keep last 100) self.persistent_notifications.append(hist_data) if len(self.persistent_notifications) > 100: - # Remove oldest notifications but keep their cached images old_notif = self.persistent_notifications.pop(0) - # Don't delete cached image here as it might be in use - # Save to file self._save_persistent_history() - logger.info( - f"Added notification to history: {hist_data['summary']} from { - hist_data['app_name'] - }" - ) + self.emit("notification-added") except Exception as e: logger.error(f"Error adding notification to history: {e}") def clear_history_for_app(self, app_name): - """Clear history for a specific app.""" try: original_count = len(self.persistent_notifications) self.persistent_notifications = [ @@ -639,9 +627,6 @@ class NotificationContainer(Box): def on_new_notification(self, fabric_notif, id): notification_history_instance = self.notification_history if notification_history_instance.do_not_disturb_enabled: - logger.info( - "Do Not Disturb mode enabled: adding notification directly to history." - ) notification = fabric_notif.get_notification_from_id(id) new_box = NotificationBox(notification) if notification.image_pixbuf: @@ -670,7 +655,6 @@ class NotificationContainer(Box): existing_notification_index ) self.stack.remove(old_notification_box) - # Add old notification to history before destroying old_notification_box.set_is_history(True) notification_history_instance.add_notification(old_notification_box) old_notification_box.destroy() @@ -712,26 +696,19 @@ class NotificationContainer(Box): self.main_revealer.set_reveal_child(True) self.update_navigation_buttons() - logger.info( - f"[NotificationContainer] New notification from {notification.app_name}" - ) - def show_previous(self, *args): - """Show the previous notification in the stack.""" if self.current_index > 0: self.current_index -= 1 self.stack.set_visible_child(self.notifications[self.current_index]) self.update_navigation_buttons() def show_next(self, *args): - """Show the next notification in the stack.""" if self.current_index < len(self.notifications) - 1: self.current_index += 1 self.stack.set_visible_child(self.notifications[self.current_index]) self.update_navigation_buttons() def update_navigation_buttons(self): - """Update navigation button states and visibility.""" self.prev_button.set_sensitive(self.current_index > 0) self.next_button.set_sensitive(self.current_index < len(self.notifications) - 1) should_reveal = len(self.notifications) > 1 @@ -744,7 +721,6 @@ class NotificationContainer(Box): return self._destroyed_notifications.add(notification.id) try: - logger.info(f"Notification {notification.id} closing with reason: {reason}") notif_to_remove = None for i, notif_box in enumerate(self.notifications): if notif_box.notification.id == notification.id: @@ -757,17 +733,10 @@ class NotificationContainer(Box): notification_history_instance = self.notification_history - # Always add to history regardless of close reason - logger.info( - f"Adding notification {notification.id} to history (reason: { - reason_str - })" - ) notif_box.set_is_history(True) notification_history_instance.add_notification(notif_box) notif_box.stop_timeout() - # Only destroy if explicitly dismissed by user if reason_str == "NotificationCloseReason.DISMISSED_BY_USER": logger.info( f"User dismissed notification { @@ -823,7 +792,6 @@ class NotificationContainer(Box): logger.error(f"Error closing notification: {e}") def _destroy_container(self): - """Clean up the container when all notifications are closed.""" try: self.notifications.clear() self._destroyed_notifications.clear() @@ -838,7 +806,6 @@ class NotificationContainer(Box): return False def pause_and_reset_all_timeouts(self): - """Pause timeouts for all notifications.""" if self._is_destroying: return for notification in self.notifications[:]: @@ -849,7 +816,6 @@ class NotificationContainer(Box): logger.error(f"Error pausing timeout: {e}") def resume_all_timeouts(self): - """Resume timeouts for all notifications.""" if self._is_destroying: return for notification in self.notifications[:]: @@ -860,7 +826,6 @@ class NotificationContainer(Box): logger.error(f"Error resuming timeout: {e}") def close_all_notifications(self, *args): - """Close all notifications.""" notifications_to_close = self.notifications.copy() for notification_box in notifications_to_close: notification_box.notification.close("dismissed-by-user") @@ -871,7 +836,6 @@ class NotificationPopup(Window): y_pos = data.NOTIF_POS.lower() if hasattr(data, "NOTIF_POS") else "top" x_pos = "right" - # Simple positioning logic - can be enhanced later if hasattr(data, "DOCK_POSITION") and data.DOCK_POSITION in ["Top", "Bottom"]: x_pos = "right" @@ -888,7 +852,9 @@ class NotificationPopup(Window): self.widgets = kwargs.get("widgets", None) self.notification_history = ( - self.widgets.notification_history if self.widgets else NotificationHistory() + self.widgets.notification_history + if self.widgets + else get_shared_notification_history() ) self.notification_container = NotificationContainer( notification_history_instance=self.notification_history, diff --git a/services/notification.py b/services/notification.py deleted file mode 100644 index 126ae3b..0000000 --- a/services/notification.py +++ /dev/null @@ -1,227 +0,0 @@ -import config.data as data -from gi.repository import GLib -import json -import os -from typing import List - -import gi -from fabric import Signal -from fabric.notifications import Notification, Notifications -from loguru import logger - -gi.require_version("GLib", "2.0") - - -NOTIFICATION_CACHE_FILE = f"{data.CACHE_DIR}/notifications.json" - - -class CustomNotifications(Notifications): - """A service to manage the notifications.""" - - @Signal - def clear_all(self, value: bool) -> None: - """Signal emitted when notifications are emptied.""" - # Implement as needed for your application - - @Signal - def notification_count(self, value: int) -> None: - """Signal emitted when a new notification is added.""" - # Implement as needed for your application - - @Signal - def dnd(self, value: bool) -> None: - """Signal emitted when dnd is toggled.""" - # Implement as needed for your application - - @property - def count(self) -> int: - """Return the count of notifications.""" - return len(self.all_notifications) - - @property - def dont_disturb(self) -> bool: - """Return the pause status.""" - return self._dont_disturb - - @dont_disturb.setter - def dont_disturb(self, value: bool): - """Set the pause status.""" - self._dont_disturb = value - self.emit("dnd", value) - - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) - self.all_notifications = [] - self._count = 0 # Will be updated to highest ID when loading - self.deserialized_notifications = [] - self._dont_disturb = False - self._load_notifications() - - def _load_notifications(self): - """Read notifications from the cache file.""" - if os.path.exists(NOTIFICATION_CACHE_FILE): - try: - with open(NOTIFICATION_CACHE_FILE, "r") as file: - notifications = json.load(file) - - def validate_with_id(notif): - """Helper to validate and return ID if valid.""" - try: - self._deserialize_notification(notif) - return (True, notif, notif.get("id", 0)) - except Exception as e: - msg = f"[Notification] Invalid: {str(e)[:50]}" - logger.error(msg) - return (False, None, 0) - - # Validate all notifications at once - results = [validate_with_id(n) for n in notifications] - - # Process results and find highest ID - valid_notifications = [] - highest_id = self._count # Start with current count - for is_valid, notif, notif_id in results: - if is_valid: - valid_notifications.append(notif) - highest_id = max(highest_id, notif_id) - - self.all_notifications = valid_notifications - self._count = highest_id # Update to highest ID seen - self._write_notifications(self.all_notifications) - - except (json.JSONDecodeError, KeyError, ValueError, IndexError) as e: - logger.error(f"[Notification] {e}") - self.all_notifications = [] - self._count = 0 - - def remove_notification(self, id: int): - """Remove the notification of given id.""" - item = next((p for p in self.all_notifications if p["id"] == id), None) - if item: - self.all_notifications.remove(item) - self._write_notifications(self.all_notifications) - self.emit("notification_count", len(self.all_notifications)) - - # Emit clear_all signal if there are no notifications left - if len(self.all_notifications) == 0: - self.emit("clear_all", True) - - def cache_notification(self, data: Notification): - """Cache the notification without widget_config and max_count.""" - - # First clean up any invalid notifications - self._cleanup_invalid_notifications() - - # Create the new notification - self._count += 1 - new_id = self._count - timestamp = GLib.get_real_time() / 1000000 # Convert microseconds to seconds - serialized_data = data.serialize() - serialized_data.update( - {"id": new_id, "app_name": data.app_name, "timestamp": timestamp} - ) - - # Append the new notification - self.all_notifications.append(serialized_data) - - self._write_notifications(self.all_notifications) - self.emit("notification_count", len(self.all_notifications)) - - def _cleanup_invalid_notifications(self): - """Remove any invalid notifications.""" - - def validate_with_id(notif): - """Helper to validate and return result with ID.""" - try: - self._deserialize_notification(notif) - return (True, notif, None) - except Exception as e: - msg = f"[Notification] Removing invalid: {str(e)[:50]}" - logger.debug(msg) - return (False, None, notif.get("id", 0)) - - # Validate all notifications at once - results = [validate_with_id(n) for n in self.all_notifications] - - # Process results - valid_notifications = [] - invalid_count = 0 - for is_valid, notif, invalid_id in results: - if is_valid: - valid_notifications.append(notif) - else: - invalid_count += 1 - self.emit("notification-closed", invalid_id, "dismissed-by-limit") - - if invalid_count > 0: - self.all_notifications = valid_notifications - self._write_notifications(self.all_notifications) - self.emit("notification_count", len(self.all_notifications)) - - def _deserialize_notification(self, notification): - """Deserialize a notification.""" - try: - notif = Notification.deserialize(notification) - # Store the timestamp in the notification object - notif.timestamp = notification.get( - "timestamp", GLib.get_real_time() / 1000000 - ) - return notif - except Exception as e: - logger.error( - f"[Notification] Deserialize failed for: {notification}, Error: {e}" - ) - raise - - def clear_all_notifications(self): - """Empty the notifications.""" - logger.info("[Notification] Clearing all notifications") - # Clear notifications but preserve the highest ID we've seen - highest_id = self._count - self.all_notifications = [] - self._write_notifications(self.all_notifications) - self.emit("notification_count", 0) - self.emit("clear_all", True) - # Restore the ID counter so new notifications get unique IDs - self._count = highest_id - - def _write_notifications(self, data): - """Write the notifications to the cache file.""" - try: - with open(NOTIFICATION_CACHE_FILE, "w") as f: - json.dump(data, f, indent=4, ensure_ascii=False) - logger.info("[Notification] Notifications written successfully.") - except Exception as e: - logger.error(f"[Notification] Failed to write notifications: {e}") - - def get_deserialized(self) -> List[Notification]: - """Return the notifications.""" - - def deserialize_with_id(notif): - """Helper to deserialize and return result with ID.""" - try: - return (self._deserialize_notification(notif), None) - except Exception as e: - msg = f"[Notification] Deserialize failed: {str(e)[:50]}" - logger.error(msg) - return (None, notif.get("id")) - - # Process all notifications at once - results = [ - deserialize_with_id(notification) for notification in self.all_notifications - ] - - # Split into successful and failed - deserialized = [] - invalid_ids = [] - for result, error_id in results: - if result is not None: - deserialized.append(result) - elif error_id is not None: - invalid_ids.append(error_id) - - # Clean up invalid notifications - for invalid_id in invalid_ids: - self.remove_notification(invalid_id) - - return deserialized diff --git a/styles/indicators.css b/styles/indicators.css index 144ca02..12a6ec6 100644 --- a/styles/indicators.css +++ b/styles/indicators.css @@ -16,7 +16,7 @@ #button-bar-recording { border-radius: 12px; - padding:4px; + padding: 4px; animation: recording-blink 1.5s ease-in-out infinite; } @@ -24,7 +24,6 @@ background-color: var(--red); } - @keyframes recording-blink { 0% { background-color: var(--red-dim); @@ -52,161 +51,3 @@ #button-bar-notifications:hover { background-color: var(--surface-bright); } - - -#notifications-popup-main { - background-color: var(--surface); - margin:20px; -} - -#notifications-popup-box { - background-color: var(--surface); -} - -#notifications-scrolled { - background-color: var(--surface); - border-radius: 8px; -} - -#notifications-scrolled scrollbar { - background-color: var(--surface-container); - border-radius: 4px; -} - -#notifications-scrolled scrollbar slider { - background-color: var(--outline); - border-radius: 4px; - min-width: 8px; - min-height: 8px; -} - -#notifications-scrolled scrollbar slider:hover { - background-color: var(--outline-variant); -} - -/* Style close button in dock notifications */ -#notifications-popup-window #notif-close-button { - background-color: var(--surface-container); - border-radius: 50%; - min-width: 28px; - min-height: 28px; - padding: 4px; - border: none; -} - -#notifications-popup-window #notif-close-button:hover { - background-color: var(--error-container); -} - -#notifications-popup-window #notif-close-label { - color: var(--on-surface-variant); - font-size: 14px; -} - -#notifications-popup-window #notif-close-button:hover #notif-close-label { - color: var(--on-error-container); -} - -#notifications-header { - padding: 8px; - border-bottom: 1px solid var(--outline); -} - -#notifications-title { - font-size: 16px; - font-weight: bold; - color: var(--on-surface); -} - -#dnd-toggle-button { - background-color: var(--surface-container); - color: var(--on-surface-variant); - border-radius: 8px; - padding: 4px 8px; - border: 1px solid var(--outline); - font-size: 12px; - font-weight: 500; -} - -#dnd-toggle-button:hover { - background-color: var(--surface-bright); -} - -#clear-all-button { - background-color: var(--primary); - color: var(--on-primary); - border-radius: 8px; - padding: 4px 8px; -} - -#clear-all-button:hover { - background-color: var(--primary-bright); -} - -#notifications-content { - padding: 8px; - background-color: var(--surface-container); - border-radius: 8px; - margin-top: 4px; - min-height: 200px; -} - -#no-notifications-content { - background-color: var(--surface-container); - border-radius: 8px; - padding: 40px 20px; - margin: 20px; - min-width: 292px; -} - -#no-notifications-icon { - font-size: 48px; - color: var(--on-surface-variant); - opacity: 0.6; -} - -#no-notifications-message { - color: var(--on-surface-variant); - font-size: 16px; - font-weight: 500; -} - -#simple-notification-box { - background-color: var(--surface-bright); - border-radius: 8px; - padding: 8px; - margin: 4px 0; -} - -#simple-notification-box:hover { - background-color: var(--surface-dim); -} - -#notification-remove-button { - background-color: var(--error); - color: var(--on-error); - border-radius: 50%; - padding: 2px 6px; - font-weight: bold; -} - -#notification-remove-button:hover { - background-color: var(--error-bright); -} - -#no-notifications-content { - padding: 48px 32px; - min-height: 200px; -} - -#no-notifications-icon { - font-size: 48px; - color: var(--outline); - opacity: 0.6; -} - -#no-notifications-message { - font-size: 16px; - color: var(--on-surface); - opacity: 0.7; -} diff --git a/styles/notification.css b/styles/notification.css index 8d7ee14..02a2584 100644 --- a/styles/notification.css +++ b/styles/notification.css @@ -38,6 +38,19 @@ margin-top: 8px; } +#no-notif { + font-size: 96px; + color: var(--surface); +} + +#notification-history { + border-radius: 20px; + padding: 16px; + border: 2px solid var(--surface); + background-color: var(--shadow); + margin: 4px; + min-width: 430px; +} #notification-stack-box { border-radius: 32px; padding: 16px; @@ -168,6 +181,37 @@ margin-left: 4px; } +#dnd-switch { + min-width: 40px; + min-height: 20px; + background-color: var(--surface); + border-radius: 15px; + padding: 2px; + transition: background-color 0.3s ease; +} + +#dnd-switch slider { + background-color: var(--primary); + border-radius: 16px; + min-width: 16px; + min-height: 8px; + transition: + background-color 0.1s cubic-bezier(0.5, 0.25, 0, 1.25), + transform 0.25s cubic-bezier(0.5, 0.25, 0, 1.25); +} + +#dnd-switch:checked { + background-color: var(--primary); +} + +#dnd-switch:checked slider { + background-color: var(--shadow); +} + +#dnd-switch:checked image { + color: var(--shadow); +} + #notif-sep { padding: 2px; border-radius: 16px;