From b497c61fa087d7d429a5583ca150ab4b246dc9cf Mon Sep 17 00:00:00 2001 From: S4NKALP Date: Wed, 6 Aug 2025 12:27:00 +0545 Subject: [PATCH] =?UTF-8?q?=E2=99=BB=EF=B8=8F=20refactor(notification):=20?= =?UTF-8?q?improve=20notification=20caching=20and=20management?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- config/assets/dock.json | 65 +-- modules/controlcenter/main.py | 20 +- modules/notification/notification.py | 24 +- modules/notification/notification_center.py | 155 ++++--- services/custom_notification.py | 444 +++++++++----------- services/modus.py | 35 +- 6 files changed, 372 insertions(+), 371 deletions(-) diff --git a/config/assets/dock.json b/config/assets/dock.json index e6d4377..021f784 100644 --- a/config/assets/dock.json +++ b/config/assets/dock.json @@ -1,33 +1,34 @@ { - "pinned_apps": [ - { - "name": "Bulk Rename", - "display_name": "Bulk Rename", - "window_class": null, - "executable": "thunar", - "command_line": "thunar --bulk-rename %F" - }, - { - "name": "Ghostty", - "display_name": "Ghostty", - "window_class": "com.mitchellh.ghostty", - "executable": "ghostty", - "command_line": "ghostty" - }, - { - "name": "Zen Browser", - "display_name": "Zen Browser", - "window_class": "zen", - "executable": "/opt/zen-browser-bin/zen-bin", - "command_line": "/opt/zen-browser-bin/zen-bin %u" - }, - { - "name": "Visual Studio Code", - "display_name": "Visual Studio Code", - "window_class": "Code", - "executable": "/usr/bin/code", - "command_line": "/usr/bin/code %F", - "icon_name": "visual-studio-code" - } - ] -} \ No newline at end of file + "pinned_apps": [ + { + "name": "Bulk Rename", + "display_name": "Bulk Rename", + "window_class": null, + "executable": "thunar", + "command_line": "thunar --bulk-rename %F" + }, + { + "name": "Ghostty", + "display_name": "Ghostty", + "window_class": "com.mitchellh.ghostty", + "executable": "ghostty", + "command_line": "ghostty" + }, + { + "name": "Zen Browser", + "display_name": "Zen Browser", + "window_class": "zen", + "executable": "/opt/zen-browser-bin/zen-bin", + "command_line": "/opt/zen-browser-bin/zen-bin %u" + }, + { + "name": "Visual Studio Code", + "display_name": "Visual Studio Code", + "window_class": "Code", + "executable": "/usr/bin/code", + "command_line": "/usr/bin/code %F", + "icon_name": "visual-studio-code" + } + ] +} + diff --git a/modules/controlcenter/main.py b/modules/controlcenter/main.py index 986f912..a230060 100644 --- a/modules/controlcenter/main.py +++ b/modules/controlcenter/main.py @@ -248,16 +248,10 @@ class ModusControlCenter(Window): style_classes="menu", h_expand=True, children=[ - Label( - label="Display", - style_classes="title", - h_align="start" - ), + Label(label="Display", style_classes="title", h_align="start"), self.brightness_scale, Label( - label="󰖨 ", - name="brightness-widget-icon", - h_align="start" + label="󰖨 ", name="brightness-widget-icon", h_align="start" ), ], ), @@ -267,16 +261,10 @@ class ModusControlCenter(Window): style_classes="menu", h_expand=True, children=[ - Label( - label="Sound", - style_classes="title", - h_align="start" - ), + Label(label="Sound", style_classes="title", h_align="start"), self.volume_scale, Label( - label=" ", - name="brightness-widget-icon", - h_align="start" + label=" ", name="brightness-widget-icon", h_align="start" ), ], ), diff --git a/modules/notification/notification.py b/modules/notification/notification.py index 01dfc93..a623ff4 100644 --- a/modules/notification/notification.py +++ b/modules/notification/notification.py @@ -16,7 +16,7 @@ from fabric.widgets.centerbox import CenterBox from fabric.widgets.eventbox import EventBox from fabric.widgets.image import Image from fabric.widgets.label import Label -from utils.roam import modus_service, notification_service +from utils.roam import modus_service from widgets.custom_image import CustomImage from widgets.customrevealer import SlideRevealer from widgets.wayland import WaylandWindow as Window @@ -458,7 +458,8 @@ class NotificationState: class ModusNoti(Window): def __init__(self): - self._server = notification_service + self._server = modus_service.notification_service + self.notifications = Box( v_expand=True, h_expand=True, @@ -490,21 +491,12 @@ class ModusNoti(Window): def on_new_notification(self, fabric_notif, id): notification: Notification = fabric_notif.get_notification_from_id(id) - # Cache the notification to the modus service for persistence - try: - modus_service.cache_notification(notification) - logger.debug( - f"Cached notification: {notification.app_name} - {notification.summary}" - ) - except Exception as e: - logger.error(f"Failed to cache notification: {e}") - - # Check if the notification is in the "do not disturb" mode, hacky way - if self._server.dont_disturb or notification.app_name in self.ignored_apps: + # Check if the notification should be ignored completely (ignored apps) + if notification.app_name in self.ignored_apps: return - if modus_service.dont_disturb: - notification.close("dismissed-by-user") + if self._server.dont_disturb or modus_service.dont_disturb: + # Notification is already cached by the service, just don't show popup return # Clear any pending notifications in queue (except the current one being shown) @@ -528,7 +520,6 @@ class ModusNoti(Window): and self.current_notification and self.notification_queue ): - # Start hiding the current notification to make room for the new one self._start_hiding_current_notification() @@ -545,7 +536,6 @@ class ModusNoti(Window): and self.notification_state == NotificationState.SHOWING and not self.current_notification._is_closing ): - self.notification_state = NotificationState.HIDING # Force close the current notification to trigger hiding animation diff --git a/modules/notification/notification_center.py b/modules/notification/notification_center.py index d81e495..640ea50 100644 --- a/modules/notification/notification_center.py +++ b/modules/notification/notification_center.py @@ -5,16 +5,18 @@ from fabric.widgets.button import Button from fabric.widgets.label import Label from fabric.widgets.scrolledwindow import ScrolledWindow from modules.notification.notification import NotificationWidget -from utils.roam import modus_service +from services.modus import notification_service from widgets.wayland import WaylandWindow as Window from fabric.widgets.image import Image class NotificationCenterWidget(NotificationWidget): - def __init__(self, notification, notification_id, **kwargs): - self.notification_id = notification_id + def __init__(self, notification, **kwargs): + self.notification_id = notification.cache_id - super().__init__(notification, timeout_ms=0, show_close_button=True, **kwargs) + super().__init__( + notification._notification, timeout_ms=0, show_close_button=True, **kwargs + ) def create_close_button(self): close_button = Button( @@ -38,7 +40,7 @@ class NotificationCenterWidget(NotificationWidget): def _on_close_clicked(self, *args): try: - modus_service.remove_notification(self.notification_id) + notification_service.remove_cached_notification(self.notification_id) except Exception as e: logger.error(f"Error removing notification {self.notification_id}: {e}") finally: @@ -70,26 +72,36 @@ class NotificationCenter(Window): NOTIFICATION_CENTER_WIDTH = 410 self.set_size_request(NOTIFICATION_CENTER_WIDTH, 600) + notification_service.connect( + "cached-notification-added", self.on_notification_added + ) + notification_service.connect( + "cached-notification-removed", self.on_notification_removed + ) + notification_service.connect("clear-all", self.on_clear_all) + notification_service.connect("notify::count", self.on_count_changed) + main_box = Box( orientation="v", spacing=5, name="noti-center-box", ) - # Header header = Box( orientation="h", spacing=10, style="margin-bottom: 10px;", ) + header.add( Label( label="Notification Center", - h_align="center", + h_align="start", h_expand=True, style="font-size: 16px; font-weight: bold; color: #ffffff;", ) ) + main_box.add(header) self.scrolled = ScrolledWindow(h_expand=True, v_expand=True) @@ -103,79 +115,106 @@ class NotificationCenter(Window): self.scrolled.add(self.notifications_box) main_box.add(self.scrolled) + # No notifications label + self.not_found_label = Label( + label="No notifications", + h_align="center", + v_align="center", + h_expand=True, + v_expand=True, + style="color: #888888; font-style: italic; margin: 20px;", + visible=(notification_service.count == 0), + ) + main_box.add(self.not_found_label) + self.clear_all_button = Button( name="noti-clear-button", label="Clear All", on_clicked=self.clear_all_notifications, + visible=(notification_service.count > 0), ) main_box.add(self.clear_all_button) self.children = main_box - # Connect service signal - modus_service.connect( - "notification_count_changed", self._on_notification_count_changed - ) + # Load existing notifications + for cached_notification in notification_service.cached_notifications: + self.notifications_box.add( + NotificationCenterWidget(notification=cached_notification) + ) self.add_keybinding("Escape", self._on_escape_pressed) - self.connect("destroy", self._on_destroy) + def on_notification_added(self, service, cached_notification): + try: + notification_widget = NotificationCenterWidget( + notification=cached_notification + ) + # Insert at the beginning to show newest first + self.notifications_box.pack_start(notification_widget, False, False, 0) + notification_widget.show_all() + logger.debug( + f"Added notification widget for {cached_notification.app_name}" + ) + except Exception as e: + logger.error(f"Error adding notification widget: {e}") + + def on_notification_removed(self, service, cached_notification): + try: + # Find and remove the corresponding widget + for child in self.notifications_box.get_children(): + if ( + hasattr(child, "notification_id") + and child.notification_id == cached_notification.cache_id + ): + child.destroy() + break + logger.debug( + f"Removed notification widget for {cached_notification.app_name}" + ) + except Exception as e: + logger.error(f"Error removing notification widget: {e}") + + def on_clear_all(self, service): + try: + for child in self.notifications_box.get_children(): + child.destroy() + logger.debug("Cleared all notification widgets") + except Exception as e: + logger.error(f"Error clearing notification widgets: {e}") + + def on_count_changed(self, service, count=None): + current_count = notification_service.count + self.not_found_label.set_visible(current_count == 0) + self.clear_all_button.set_visible(current_count > 0) + self.scrolled.set_visible(current_count > 0) + def clear_all_notifications(self, *_): - modus_service.clear_all_notifications() - self.refresh_notifications() - self.mousecapture.hide_child_window() + notification_service.clear_all_cached_notifications() + if hasattr(self, "mousecapture"): + self.mousecapture.hide_child_window() def _on_escape_pressed(self, *_): - self.mousecapture.hide_child_window() + if hasattr(self, "mousecapture"): + self.mousecapture.hide_child_window() def _init_mousecapture(self, mousecapture): self.mousecapture = mousecapture def _set_mousecapture(self, visible): - self.refresh_notifications() - - def _on_notification_count_changed(self, *_): - self.refresh_notifications() + # No need to refresh on visibility change since we use signals + pass def _on_destroy(self, *_): - # Disconnect service signal - modus_service.disconnect( - "notification_count_changed", self._on_notification_count_changed - ) - - def refresh_notifications(self): try: - for child in self.notifications_box.get_children(): - child.destroy() - - notifications_with_ids = modus_service.get_deserialized_with_ids() - logger.info(f"Found {len(notifications_with_ids)} notifications to display") - - if not notifications_with_ids: - no_notifications_label = Label( - label="No notifications", - h_align="center", - style="color: #888888; font-style: italic; margin: 20px;", - ) - self.notifications_box.add(no_notifications_label) - return - - for notification, notification_id in notifications_with_ids: - try: - logger.debug( - f"Notification from {notification.app_name}: ID={ - notification_id - }" - ) - notification_widget = NotificationCenterWidget( - notification=notification, notification_id=notification_id - ) - self.notifications_box.add(notification_widget) - except Exception as e: - logger.error(f"Error creating notification widget: {e}") - - self.clear_all_button.set_visible(len(notifications_with_ids) > 0) - + notification_service.disconnect( + "cached-notification-added", self.on_notification_added + ) + notification_service.disconnect( + "cached-notification-removed", self.on_notification_removed + ) + notification_service.disconnect("clear-all", self.on_clear_all) + notification_service.disconnect("notify::count", self.on_count_changed) except Exception as e: - logger.error(f"Error refreshing notifications: {e}") + logger.error(f"Error disconnecting signals: {e}") diff --git a/services/custom_notification.py b/services/custom_notification.py index 6354c37..12bafa9 100644 --- a/services/custom_notification.py +++ b/services/custom_notification.py @@ -1,52 +1,149 @@ import json import os -import threading -from typing import Dict, List +from typing import List -from fabric import Signal -from fabric.notifications import Notification, Notifications, NotificationSerializedData -from loguru import logger +import gi +from gi.repository import GdkPixbuf import config.data as data +from fabric.core.service import Property, Service, Signal +from fabric.notifications import ( + Notification, + NotificationAction, + NotificationImagePixmap, + Notifications, +) -PERSISTENT_DIR = f"/tmp/{data.APP_NAME}/notifications" -NOTIFICATION_CACHE_FILE = os.path.join(PERSISTENT_DIR, "notification_history.json") +gi.require_version("Gtk", "3.0") + +NOTIFICATION_CACHE_FILE = f"{data.CACHE_DIR}/notification" -def write_json_file(data: Dict, path: str): - try: - # Create directory if it doesn't exist - os.makedirs(os.path.dirname(path), exist_ok=True) - with open(path, "w") as f: - json.dump(data, f, indent=4, ensure_ascii=False) - except Exception as e: - logger.warning(f"Failed to write json: {e}") - - -class CustomNotifications(Notifications): - """A service to manage the notifications.""" +class CachedNotification(Service): + @classmethod + def create_from_dict(cls, data, **kwargs): + data["timeout"] = 0 + self = cls.__new__(cls) + Service.__init__(self, **kwargs) + self._notification = Notification.deserialize(data) + self.cache_id = data["cached-id"] + return self @Signal - def clear_all(self, value: bool) -> None: + def removed_from_cache(self) -> None: ... + + @Property(int, "read-write") + def cache_id(self) -> int: + return self._cache_id + + @cache_id.setter + def cache_id(self, cache_id: int): + self._cache_id = cache_id + + @Property(str, "readable") + def app_name(self) -> str: + return self._notification.app_name + + @Property(str, "readable") + def app_icon(self) -> str: + return self._notification.app_icon + + @Property(str, "readable") + def summary(self) -> str: + return self._notification.summary + + @Property(str, "readable") + def body(self) -> str: + return self._notification.body + + @Property(int, "readable") + def id(self) -> int: + return self._notification.id + + @Property(int, "readable") + def replaces_id(self) -> int: + return self._notification.replaces_id + + @Property(int, "readable") + def urgency(self) -> int: + return self._notification.urgency + + @Property(list[NotificationAction], "readable") + def actions(self) -> list[NotificationAction]: + return self._notification.actions + + @Property(NotificationImagePixmap, "readable") + def image_pixmap(self) -> NotificationImagePixmap: + return self._notification.image_pixmap # type: ignore + + @Property(str, "readable") + def image_file(self) -> str: + return self._notification.image_file # type: ignore + + @Property(GdkPixbuf.Pixbuf, "readable") + def image_pixbuf(self) -> GdkPixbuf.Pixbuf: + if self.image_pixmap: + return self.image_pixmap.as_pixbuf() + if self.image_file: + return GdkPixbuf.Pixbuf.new_from_file(self.image_file) + return None # type: ignore + + @Property(dict, "readable") + def serialized(self) -> dict: + return { + "cached-id": self.cache_id, + "id": self.id, + "replaces-id": self.replaces_id, + "app-name": self.app_name, + "app-icon": self.app_icon, + "summary": self.summary, + "body": self.body, + "urgency": self.urgency, + "actions": [(action.identifier, action.label) for action in self.actions], + "image-file": self.image_file, + "image-pixmap": ( + self.image_pixmap.serialize() if self.image_pixmap else None + ), + } + + def __init__(self, notification: Notification, cache_id: int, **kwargs): + super().__init__() + self._notification: Notification = notification + self._cache_id = cache_id + + def remove_from_cache(self): + self.removed_from_cache.emit() + + +class CachedNotifications(Notifications): + """A service to manage the cached notifications.""" + + @Signal + def clear_all(self) -> None: """Signal emitted when notifications are emptied.""" - # Implement as needed for your application + pass @Signal - def notification_count(self, value: int) -> None: - """Signal emitted when a new notification is added.""" - # Implement as needed for your application + def cached_notification_added(self, notification: CachedNotification) -> None: + """Signal emitted when a notification is cached.""" + pass @Signal - def dnd(self, value: bool) -> None: - """Signal emitted when dnd is toggled.""" - # Implement as needed for your application + def cached_notification_removed(self, notification: CachedNotification) -> None: + """Signal emitted when a notification is removed from cache.""" + pass - @property + @Property(List[CachedNotification], "readable") + def cached_notifications(self) -> List[CachedNotification]: + """Return the cached notifications.""" + return list(self._cached_notifications.values()) + + @Property(int, "readable") def count(self) -> int: """Return the count of notifications.""" - return len(self.all_notifications) + return self._count - @property + @Property(bool, "read-write", default_value=False) def dont_disturb(self) -> bool: """Return the pause status.""" return self._dont_disturb @@ -55,227 +152,106 @@ class CustomNotifications(Notifications): def dont_disturb(self, value: bool): """Set the pause status.""" self._dont_disturb = value - self.emit("dnd", value) + self.notify("dont-disturb") - 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 = [] + def __init__(self, **kwargs): + super().__init__() + self._cached_notifications: dict[int, CachedNotification] = {} + self._signal_handlers = {} # Store signal handlers by notification_id self._dont_disturb = False - self._lock = threading.Lock() # Add missing lock for thread safety - self._load_notifications() + self._count = 0 - def _load_notifications(self): - """Read and validate notifications from the cache file.""" - # Create directory if it doesn't exist - os.makedirs(os.path.dirname(NOTIFICATION_CACHE_FILE), exist_ok=True) - - if not os.path.exists(NOTIFICATION_CACHE_FILE): - return + self.load_cached_notifications() + def load_cached_notifications(self) -> dict[int, CachedNotification]: + """Load cached notifications from a JSON file (deserialization).""" try: with open(NOTIFICATION_CACHE_FILE, "r") as file: - original_data = json.load(file) + data = json.load(file) # Load list of serialized notifications + except (FileNotFoundError, json.JSONDecodeError): + # If file doesn't exist or is corrupted, start with empty list + data = [] - original_data.reverse() + for notification in data: + cached_notification = CachedNotification.create_from_dict(notification) + handler_id = cached_notification.connect( + "removed-from-cache", + lambda *args: self.remove_cached_notification( + notification_id=cached_notification.cache_id + ), + ) + self._signal_handlers[cached_notification.cache_id] = handler_id + self._cached_notifications[cached_notification.cache_id] = ( + cached_notification + ) + self._count += 1 - valid_notifications = [] - highest_id = self._count + self.notify("count") - for notification in original_data: - try: - self._deserialize_notification(notification) - valid_notifications.append(notification) - highest_id = max(highest_id, notification.get("id", 0)) - except Exception as e: - msg = f"[Notification] Invalid: {str(e)[:50]}" - logger.exception(f"{msg}") + def cache_notifications(self) -> None: + """Save cached notifications to a JSON file.""" + # Ensure cache directory exists + os.makedirs(os.path.dirname(NOTIFICATION_CACHE_FILE), exist_ok=True) - # Write only if the validated data differs from what was originally loaded - if valid_notifications != original_data: - write_json_file(valid_notifications, NOTIFICATION_CACHE_FILE) - logger.info("[Notification] Notifications written successfully.") + serialized_data = [ + notif.serialized for notif in self._cached_notifications.values() + ] # Convert to serializable format + with open(NOTIFICATION_CACHE_FILE, "w") as file: + json.dump(serialized_data, file, indent=4) - self.all_notifications = valid_notifications - self._count = highest_id - - del valid_notifications - del original_data - - except (json.JSONDecodeError, KeyError, ValueError, IndexError) as e: - logger.exception(f"[Notification] {e}") - - def remove_notification(self, id: int): - """Remove a notification by ID, ensuring thread safety.""" - with self._lock: - item = next((p for p in self.all_notifications if p["id"] == id), None) - if item: - self.all_notifications.remove(item) - self._persist_and_emit() - - if len(self.all_notifications) == 0: - self.emit("clear_all", True) - - def cache_notification(self, widget_config, data: Notification, max_count: int): - """Cache a notification, ensuring thread safety.""" - with self._lock: - self._cleanup_invalid_notifications() - new_notification = self._create_serialized_notification(data) - self._enforce_per_app_limit(widget_config, new_notification, max_count) - self.all_notifications.append(new_notification) - self._enforce_global_limit(max_count) - self._persist_and_emit() - - def _cleanup_invalid_notifications(self): - """Remove any invalid notifications.""" - - valid_notifications = [] - invalid_count = 0 - - for notification in self.all_notifications: - try: - self._deserialize_notification(notification) - valid_notifications.append(notification) - except Exception as e: - msg = f"[Notification] Removing invalid: {str(e)[:50]}" - logger.debug(msg) - invalid_id = notification.get("id", 0) - self.emit("notification-closed", invalid_id, "dismissed-by-limit") - invalid_count += 1 - - if invalid_count > 0: - self.all_notifications = valid_notifications - self._persist_and_emit() - del valid_notifications - logger.info(f"[Notification] Cleaned {invalid_count} invalid notifications") - - def _create_serialized_notification(self, data: Notification) -> dict: - """Generate a new notification with a unique ID.""" - self._count += 1 - serialized = data.serialize() - serialized.update( - { - "id": self._count, - "app-name": data.app_name, - } - ) - return serialized - - def _enforce_global_limit(self, max_count: int): - """Remove oldest notifications if total count exceeds global limit.""" - while len(self.all_notifications) > max_count: - oldest = self.all_notifications.pop(0) - self.emit("notification-closed", oldest["id"], "dismissed-by-limit") - - def _enforce_per_app_limit( - self, widget_config, new_notification: dict, max_count: int - ): - """Ensure per-app limits are respected.""" - app_name = new_notification["app-name"] - per_app_limits = widget_config.get("notification", {}).get("per_app_limits", {}) - app_limit = per_app_limits.get(app_name, max_count) - - app_notifications = [ - n for n in self.all_notifications if n["app-name"] == app_name - ] - - if len(app_notifications) >= app_limit: - app_notifications.sort(key=lambda x: x["id"]) # Oldest first - to_remove = len(app_notifications) - app_limit + 1 - for old in app_notifications[:to_remove]: - self.all_notifications.remove(old) - self.emit("notification-closed", old["id"], "dismissed-by-limit") - - def _deserialize_notification(self, notification: NotificationSerializedData): - """Deserialize a notification.""" - return Notification.deserialize(notification) - - def _persist_and_emit(self): - """Persist notifications and emit relevant signals.""" - write_json_file(self.all_notifications, NOTIFICATION_CACHE_FILE) - self.emit("notification_count", len(self.all_notifications)) - - def clear_all_notifications(self): + def clear_all_cached_notifications(self): """Empty the notifications.""" - logger.info("[Notification] Clearing all notifications") + for cached_notification in self._cached_notifications.values(): + handler_id = self._signal_handlers.pop(cached_notification.cache_id, None) + if handler_id: + cached_notification.disconnect(handler_id) + self._cached_notifications = {} + self.cache_notifications() + self._count = 0 + self.notify("count") + self.clear_all.emit() - # Clear notifications but preserve the highest ID we've seen - highest_id = self._count + def notification_added(self, notification_id: int) -> None: + """Handle notification added and cache it.""" + super().notification_added(notification_id) - self.all_notifications.clear() + notification = self.get_notification_from_id(notification_id) - self._persist_and_emit() + if notification: + # Always cache notifications, regardless of DND status + # DND only affects popup display, not caching + self._count += 1 + cache_id = self._count - logger.info("[Notification] Notifications written successfully.") + cached_notification = CachedNotification( + notification=notification, cache_id=cache_id + ) + handler_id = cached_notification.connect( + "removed-from-cache", + lambda *args: self.remove_cached_notification(notification_id=cache_id), + ) - self.emit("clear_all", True) + self._signal_handlers[cache_id] = handler_id + self._cached_notifications[cache_id] = cached_notification + self.cache_notifications() - # Restore the ID counter so new notifications get unique IDs - self._count = highest_id + self.notify("count") + self.emit("cached-notification-added", cached_notification) - def get_deserialized(self) -> List[Notification]: - """Return the notifications.""" + def remove_cached_notification(self, notification_id: int): + """Remove the notification of given id.""" + if notification_id in self._cached_notifications: + cached_notification = self._cached_notifications.pop( + notification_id + ) # Remove from cache + self.cache_notifications() # Update JSON + self._count -= 1 + self.notify("count") + # Get the stored signal handler ID and disconnect it + handler_id = self._signal_handlers.pop(notification_id, None) + if handler_id: + # Disconnect the signal handler + cached_notification.disconnect(handler_id) - def deserialize_with_id(notification): - """Helper to deserialize and return result with ID.""" - try: - return (self._deserialize_notification(notification), None) - except Exception as e: - msg = f"[Notification] Deserialize failed: {str(e)[:50]}" - logger.exception(f"{msg}") - return (None, notification.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 - - def get_deserialized_with_ids(self) -> List[tuple[Notification, int]]: - """Return the notifications with their IDs as tuples.""" - - def deserialize_with_id(notification): - """Helper to deserialize and return result with ID.""" - try: - deserialized = self._deserialize_notification(notification) - notification_id = notification.get("id") - return (deserialized, notification_id, None) - except Exception as e: - msg = f"[Notification] Deserialize failed: {str(e)[:50]}" - logger.exception(f"{msg}") - return (None, notification.get("id"), notification.get("id")) - - # Process all notifications at once - results = [ - deserialize_with_id(notification) for notification in self.all_notifications - ] - - # Split into successful and failed - deserialized_with_ids = [] - invalid_ids = [] - for result, notification_id, error_id in results: - if result is not None: - deserialized_with_ids.append((result, notification_id)) - 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_with_ids + def toggle_dnd(self): + self.dont_disturb = not self.dont_disturb diff --git a/services/modus.py b/services/modus.py index 9e9ada8..add386d 100644 --- a/services/modus.py +++ b/services/modus.py @@ -6,9 +6,10 @@ from loguru import logger from fabric.core.service import Property, Service, Signal from fabric.hyprland.service import Hyprland from fabric.notifications import Notification -from services.custom_notification import CustomNotifications +from services.new_notification import CachedNotifications + +notification_service = CachedNotifications() -notification_service = CustomNotifications() class ModusService(Service): @Signal @@ -251,12 +252,16 @@ class ModusService(Service): self._hyprland_connection = Hyprland() # Get initial workspace - workspace_data = self._hyprland_connection.send_command("j/activeworkspace").reply + workspace_data = self._hyprland_connection.send_command( + "j/activeworkspace" + ).reply active_workspace = json.loads(workspace_data.decode("utf-8"))["name"] self._current_workspace = str(active_workspace) # Connect to workspace change events - self._hyprland_connection.connect("event::workspace", self._on_workspace_changed) + self._hyprland_connection.connect( + "event::workspace", self._on_workspace_changed + ) except Exception as e: logger.error(f"[ModusService] Failed to setup workspace monitoring: {e}") @@ -271,28 +276,30 @@ class ModusService(Service): logger.error(f"[ModusService] Error processing workspace change: {e}") def remove_notification(self, id: int): - notification_service.remove_notification(id) - self.notification_count_changed(self.notification_count) - - def cache_notification(self, data: Notification): - widget_config = {"notification": {"per_app_limits": {}}} - notification_service.cache_notification(widget_config, data, max_count=100) + notification_service.remove_cached_notification(id) self.notification_count_changed(self.notification_count) def clear_all_notifications(self): - notification_service.clear_all_notifications() + notification_service.clear_all_cached_notifications() self.notification_count_changed(self.notification_count) - def get_deserialized(self) -> List[Notification]: - return notification_service.get_deserialized() + def get_cached_notifications(self): + return notification_service.cached_notifications def get_deserialized_with_ids(self): - return notification_service.get_deserialized_with_ids() + return [ + (notif._notification, notif.cache_id) + for notif in notification_service.cached_notifications + ] @property def notification_count(self) -> int: return notification_service.count + def toggle_dnd(self): + notification_service.toggle_dnd() + self.dont_disturb_changed(notification_service.dont_disturb) + global modus_service try: