sankalp/Modus
1
0 Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases 1 Wiki Activity

♻️ refactor(notification): improve notification caching and management

Browse Source
This commit is contained in:
sankalp committed 2025-08-06 12:27:00 +05:45
1 parent c1c6786b26
commit b497c61fa0
6 files changed
+372 -371
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

config/assets/dock.json
+33 -32
View File
@@ -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"
}
]
}
"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"
}
]
}
modules/controlcenter/main.py
+4 -16
View File
@@ -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"
),
],
),
modules/notification/notification.py
+7 -17
View File
@@ -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
modules/notification/notification_center.py
+97 -58
View File
@@ -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}")
services/custom_notification.py
+210 -234
View File
@@ -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
services/modus.py
+21 -14
View File
@@ -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: